Disclaimer

This blog post is a part of NSA Codebreaker 2024 writeup.

The challenge content is a PURELY FICTIONAL SCENARIO created by the NSA for EDUCATIONAL PURPOSES only. The mention and use of any actual products, tools, and techniques are similarly contrived for the sake of the challenge alone, and do not represent the intent of any company, product owner, or standards body.

Any similarities to real persons, entities, or events is coincidental.

Synopsis

Great work finding those files! Barry shares the files you extracted with the blue team who share it back to Aaliyah and her team. As a first step, she ran strings across all the files found and noticed a reference to a known DIB, “Guardian Armaments” She begins connecting some dots and wonders if there is a connection between the software and the hardware tokens. But what is it used for and is there a viable threat to Guardian Armaments (GA)?

She knows the Malware Reverse Engineers are experts at taking software apart and figuring out what it’s doing. Aaliyah reaches out to them and keeps you in the loop. Looking at the email, you realize your friend Ceylan is touring on that team! She is on her first tour of the Computer Network Operations Development Program

Barry opens up a group chat with three of you. He wants to see the outcome of the work you two have already contributed to. Ceylan shares her screen with you as she begins to reverse the software. You and Barry grab some coffee and knuckle down to help.

Figure out how the APT would use this software to their benefit

Downloads

Executable from ZFS filesystem (server)
Retrieved from the facility, could be important? (shredded.jpg)

Prompt

Enter a valid JSON that contains the (3 interesting) keys and specific values that would have been logged if you had successfully leveraged the running software. Do ALL your work in lower case.

Solution

I downloaded the attachments and start poking the files. Based from the error, it seems like it connects to a server or some sort.

Upon digging more, it seems like this is an application with protobuf definitions.

Now, we will extract protobuf definitions using https://github.com/arkadiyt/protodump

Since we have now the protobuf definition, we can now create a server simulator to observe the behavior and dig more deep in the application. We can use https://pypi.org/project/grpcio-tools/

We can now write our server simulator.

from concurrent import futures
import time
import grpc
import logging

import auth_pb2
import auth_pb2_grpc

_ONE_DAY_IN_SECONDS = 60 * 60 * 24

# Configure logging
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

# Interceptor for handling logging on method not found or decoding errors
class LoggingInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        method = handler_call_details.method
        logging.info(f"Incoming request for method: {method}")
        
        try:
            # Call the actual service method
            response = continuation(handler_call_details)
            return response
        except grpc.RpcError as e:
            # Log error details
            if e.code() == grpc.StatusCode.UNIMPLEMENTED:
                logging.error(f"Method not found: {method}")
            elif e.code() == grpc.StatusCode.INVALID_ARGUMENT:
                logging.error(f"Request decoding error for method: {method}")
            else:
                logging.error(f"Error during request handling: {e}")
            raise e

class AuthService(auth_pb2_grpc.AuthServiceServicer):
    def log_metadata(self, context):
        # Log the incoming metadata (headers)
        metadata = context.invocation_metadata()
        logging.info("Received headers:")
        for key, value in metadata:
            logging.info(f"{key}: {value}")

    def Ping(self, request, context):
        # Log headers and request details
        self.log_metadata(context)
        logging.debug(f"Received Ping request: {request}")
        
        # Return the Ping response
        return auth_pb2.PingResponse(response=1)

    def Authenticate(self, request, context):
        # Log headers and request details
        self.log_metadata(context)
        logging.debug(f"Received Authenticate request: {request}")
        
        # Return the Authenticate response
        return auth_pb2.AuthResponse(success=True)

    def RegisterOTPSeed(self, request, context):
        # Log headers and request details
        self.log_metadata(context)
        logging.debug(f"Received RegisterOTPSeed request: {request}")
        
        # Return the RegisterOTPSeed response
        return auth_pb2.RegisterOTPSeedResponse(success=False)

    def VerifyOTP(self, request, context):
        # Log headers and request details
        self.log_metadata(context)
        logging.debug(f"Received VerifyOTP request: {request}")
        
        # Return the RegisterOTPSeed response
        return auth_pb2.VerifyOTPResponse(success=True,token="000000")

def serve():
    # Add the interceptor to the server
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        interceptors=[LoggingInterceptor()]
    )
    
    auth_pb2_grpc.add_AuthServiceServicer_to_server(AuthService(), server)
    server.add_insecure_port("[::]:50052")
    server.start()

    # Log the server start event
    logging.info("gRPC server started on port 50052")

    try:
        while True:
            time.sleep(_ONE_DAY_IN_SECONDS)
    except KeyboardInterrupt:
        # Log the server stop event
        logging.info("Stopping gRPC server...")
        server.stop(grace=0)
        logging.info("gRPC server stopped")

if __name__ == "__main__":
    serve()

We then also create a script for a client to connect to the server running at 50051.

import argparse

import grpc

import seed_generation_pb2
import seed_generation_pb2_grpc

def run(host):
    channel = grpc.insecure_channel(host)
    stub = seed_generation_pb2_grpc.SeedGenerationServiceStub(channel)

    response = stub.GetSeed(seed_generation_pb2.GetSeedRequest(username="jasper_05376",password="test"))
    print("SeedGenerationService client received: Seed=" + str(response.seed) + ", Count=" + str(response.count))

if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument("--host", default="localhost:50051", help="The server host.")
    args = parser.parse_args()
    run(args.host)

Based on the context clues given, it seems like we need to submit a json that has the following keys: username, seed, and count.

Upon digging more, it seems like the seeds are deterministic, so there must be a fixed seed used in instantiation.

Another piece of information is the auth module.

The code above is not safe. The application authenticates the test but not in a safe manner.
Therefore, if we can produce a combination of username and a seed that would satisfy the conditions, we might get authenticated without the need of Auth Service.

Here is the snippet of the algorithm used by the application:

v7 = currentRand;
  for ( i = 0LL; username.len > (__int64)i; i += 4LL )
  {
    if ( username.len < (__int64)(i + 4) )
    {
      v10 = username.len - i;
      if ( username.len - i == 1 )
      {
        if ( username.len <= i )
          runtime_panicIndex();
        v9 = username.str[i];
      }
      else if ( v10 == 2 )
      {
        if ( username.len <= i )
          runtime_panicIndex();
        if ( username.len <= i + 1 )
          runtime_panicIndex();
        v9 = *(unsigned __int16 *)&username.str[i];
      }
      else if ( v10 == 3 )
      {
        if ( username.len <= i )
          runtime_panicIndex();
        if ( username.len <= i + 1 )
          runtime_panicIndex();
        if ( username.len <= i + 2 )
          runtime_panicIndex();
        v9 = *(unsigned __int16 *)&username.str[i] | (username.str[i + 2] << 16);
      }
      else
      {
        v9 = 0;
      }
    }
    else
    {
      if ( username.len <= i )
        runtime_panicIndex();
      if ( username.len <= i + 1 )
        runtime_panicIndex();
      if ( username.len <= i + 2 )
        runtime_panicIndex();
      if ( username.len <= i + 3 )
        runtime_panicIndex();
      v9 = *(_DWORD *)&username.str[i];
    }
    v7 ^= v9;
  }
  if ( v7 == -1972368894 )
  {
    // we need to get into these block by using a VALID and KNOWN username.
  }

Another context clue is the shredded.jpg file. With those context clues, we can try using jasper_05376 that was found from Task 1.

With all of the information above, we can try to emulate the algorithm to bruteforce a valid username and seed combination that would meet our goal.

package main

import (
	"encoding/binary"
	"fmt"
	"math/rand"
)

// getChunks splits a username into 4-byte chunks for XOR operations
func getChunks(username string) []uint32 {
	usernameBytes := []byte(username)
	padding := (4 - len(usernameBytes)%4) % 4
	usernameBytes = append(usernameBytes, make([]byte, padding)...)

	var chunks []uint32
	for i := 0; i < len(usernameBytes); i += 4 {
		chunks = append(chunks, binary.LittleEndian.Uint32(usernameBytes[i:i+4]))
	}
	return chunks
}

// performXOR performs XOR on the initial uVar2 with chunks from the username
func performXOR(uVar2 uint32, chunks []uint32) uint32 {
	for _, chunk := range chunks {
		uVar2 ^= chunk
	}
	return uVar2
}

// simulateAuthBypass checks if the current random value XOR'd with the username meets the bypass condition
func simulateAuthBypass(username string, uVar2Initial uint32, targetUVar2 uint32) (bool, uint32) {
	usernameChunks := getChunks(username)
	finalUVar2 := performXOR(uVar2Initial, usernameChunks)

	// Return true if the bypass condition is met
	if finalUVar2 == targetUVar2 {
		return true, finalUVar2
	}
	return false, finalUVar2
}

func main() {
	// Fixed initial random seed from the server code
	seed := int64(0x76546CC2CA2D7)
	rand.Seed(seed)

	// Bypass target value
	targetUVar2 := uint32(0x8A700A02)

	// Iterate over seed count (1,000,000,000,000,000 iterations)
	maxIterations := 1000000000000000

	username := "jasper_05376"

	fmt.Println("Starting bypass detection...")

	prevRand := int64(0)

	showNextRand := false

	for count := 1; count <= maxIterations; count++ {
		// Get the current random value (uVar2Initial is the lower 32-bits of currentRand)
		currentRand := rand.Int63()
		uVar2Initial := uint32(currentRand & 0xFFFFFFFF)

		bypass, finalUVar2 := simulateAuthBypass(username, uVar2Initial, targetUVar2)

		// this will be hit on next loop after finding the bypass
		if showNextRand {
			showNextRand = false
			fmt.Printf("NextRand: %d\n\n", currentRand)
		}

		if bypass {
			// Display the bypass information
			fmt.Printf("\nBypass detected!\n")
			fmt.Printf("Username: %s\n", username)
			fmt.Printf("Seed Count: %d\n", count)
			fmt.Printf("Initial uVar2: 0x%x\n", uVar2Initial)
			fmt.Printf("Final uVar2: 0x%x (Matches Bypass Value)\n", finalUVar2)
			fmt.Printf("CurrentRand: %d\n\n", currentRand)
			fmt.Printf("PrevRand: %d\n\n", prevRand)
			showNextRand = true
		}

		prevRand = currentRand

		// Show progress
		if count%100000000 == 0 {
			fmt.Printf("Processed %d seed iterations...\n", count)
		}
	}

	fmt.Println("Bypass detection completed.")
}

Gotcha! We have now our bypass!

username: jasper_05376
count: 181182686
seed: 350024956464939860

I am not really sure why NextRand should be the value of seed key in the json, when I try submitting the CurrentRand, it just don’t accept my answer. I also forgot if its -+1 Seed Count, I was just doing mix and match with the answers until it has been accepted by the system. It’s kinda confusing but it is what it is.

In real life, I think this is an exploit that is a very hard to do. Because the attacker must track the current seed and count of the target server, and the payload must be sent in a pixel perfect timing.