#!/usr/bin/python # Copyright 2012-2017, The Tor Project, Inc # See LICENSE for licensing information """ ntor_ref.py This module is a reference implementation for the "ntor" protocol s proposed by Goldberg, Stebila, and Ustaoglu and as instantiated in Tor Proposal 216. It's meant to be used to validate Tor's ntor implementation. It requirs the curve25519 python module from the curve25519-donna package. *** DO NOT USE THIS IN PRODUCTION. *** commands: gen_kdf_vectors: Print out some test vectors for the RFC5869 KDF. timing: Print a little timing information about this implementation's handshake. self-test: Try handshaking with ourself; make sure we can. test-tor: Handshake with tor's ntor implementation via the program src/test/test-ntor-cl; make sure we can. """ import binascii try: import curve25519 curve25519mod = curve25519.keys except ImportError: curve25519 = None import slownacl_curve25519 curve25519mod = slownacl_curve25519 import hashlib import hmac import subprocess import sys # ********************************************************************** # Helpers and constants def HMAC(key,msg): "Return the HMAC-SHA256 of 'msg' using the key 'key'." H = hmac.new(key, b"", hashlib.sha256) H.update(msg) return H.digest() def H(msg,tweak): """Return the hash of 'msg' using tweak 'tweak'. (In this version of ntor, the tweaked hash is just HMAC with the tweak as the key.)""" return HMAC(key=tweak, msg=msg) def keyid(k): """Return the 32-byte key ID of a public key 'k'. (Since we're using curve25519, we let k be its own keyid.) """ return k.serialize() NODE_ID_LENGTH = 20 KEYID_LENGTH = 32 G_LENGTH = 32 H_LENGTH = 32 PROTOID = b"ntor-curve25519-sha256-1" M_EXPAND = PROTOID + b":key_expand" T_MAC = PROTOID + b":mac" T_KEY = PROTOID + b":key_extract" T_VERIFY = PROTOID + b":verify" def H_mac(msg): return H(msg, tweak=T_MAC) def H_verify(msg): return H(msg, tweak=T_VERIFY) class PrivateKey(curve25519mod.Private): """As curve25519mod.Private, but doesn't regenerate its public key every time you ask for it. """ def __init__(self): curve25519mod.Private.__init__(self) self._memo_public = None def get_public(self): if self._memo_public is None: self._memo_public = curve25519mod.Private.get_public(self) return self._memo_public # ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ if sys.version < '3': def int2byte(i): return chr(i) else: def int2byte(i): return bytes([i]) def kdf_rfc5869(key, salt, info, n): prk = HMAC(key=salt, msg=key) out = b"" last = b"" i = 1 while len(out) < n: m = last + info + int2byte(i) last = h = HMAC(key=prk, msg=m) out += h i = i + 1 return out[:n] def kdf_ntor(key, n): return kdf_rfc5869(key, T_KEY, M_EXPAND, n) # ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ def client_part1(node_id, pubkey_B): """Initial handshake, client side. From the specification: <> Takes node_id -- a digest of the server's identity key, pubkey_B -- a public key for the server. Returns a tuple of (client secret key x, client->server message)""" assert len(node_id) == NODE_ID_LENGTH key_id = keyid(pubkey_B) seckey_x = PrivateKey() pubkey_X = seckey_x.get_public().serialize() message = node_id + key_id + pubkey_X assert len(message) == NODE_ID_LENGTH + H_LENGTH + H_LENGTH return seckey_x , message def hash_nil(x): """Identity function: if we don't pass a hash function that does nothing, the curve25519 python lib will try to sha256 it for us.""" return x def bad_result(r): """Helper: given a result of multiplying a public key by a private key, return True iff one of the inputs was broken""" assert len(r) == 32 return r == '\x00'*32 def server(seckey_b, my_node_id, message, keyBytes=72): """Handshake step 2, server side. From the spec: << The server generates a keypair of y,Y = KEYGEN(), and computes secret_input = EXP(X,y) | EXP(X,b) | ID | B | X | Y | PROTOID KEY_SEED = H(secret_input, t_key) verify = H(secret_input, t_verify) auth_input = verify | ID | B | Y | X | PROTOID | "Server" The server sends a CREATED cell containing: SERVER_PK: Y -- G_LENGTH bytes AUTH: H(auth_input, t_mac) -- H_LENGTH byets >> Takes seckey_b -- the server's secret key my_node_id -- the servers's public key digest, message -- a message from a client keybytes -- amount of key material to generate Returns a tuple of (key material, sever->client reply), or None on error. """ assert len(message) == NODE_ID_LENGTH + H_LENGTH + H_LENGTH if my_node_id != message[:NODE_ID_LENGTH]: return None badness = (keyid(seckey_b.get_public()) != message[NODE_ID_LENGTH:NODE_ID_LENGTH+H_LENGTH]) pubkey_X = curve25519mod.Public(message[NODE_ID_LENGTH+H_LENGTH:]) seckey_y = PrivateKey() pubkey_Y = seckey_y.get_public() pubkey_B = seckey_b.get_public() xy = seckey_y.get_shared_key(pubkey_X, hash_nil) xb = seckey_b.get_shared_key(pubkey_X, hash_nil) # secret_input = EXP(X,y) | EXP(X,b) | ID | B | X | Y | PROTOID secret_input = (xy + xb + my_node_id + pubkey_B.serialize() + pubkey_X.serialize() + pubkey_Y.serialize() + PROTOID) verify = H_verify(secret_input) # auth_input = verify | ID | B | Y | X | PROTOID | "Server" auth_input = (verify + my_node_id + pubkey_B.serialize() + pubkey_Y.serialize() + pubkey_X.serialize() + PROTOID + b"Server") msg = pubkey_Y.serialize() + H_mac(auth_input) badness += bad_result(xb) badness += bad_result(xy) if badness: return None keys = kdf_ntor(secret_input, keyBytes) return keys, msg def client_part2(seckey_x, msg, node_id, pubkey_B, keyBytes=72): """Handshake step 3: client side again. From the spec: << The client then checks Y is in G^* [see NOTE below], and computes secret_input = EXP(Y,x) | EXP(B,x) | ID | B | X | Y | PROTOID KEY_SEED = H(secret_input, t_key) verify = H(secret_input, t_verify) auth_input = verify | ID | B | Y | X | PROTOID | "Server" The client verifies that AUTH == H(auth_input, t_mac). >> Takes seckey_x -- the secret key we generated in step 1. msg -- the message from the server. node_id -- the node_id we used in step 1. server_key -- the same public key we used in step 1. keyBytes -- the number of bytes we want to generate Returns key material, or None on error """ assert len(msg) == G_LENGTH + H_LENGTH pubkey_Y = curve25519mod.Public(msg[:G_LENGTH]) their_auth = msg[G_LENGTH:] pubkey_X = seckey_x.get_public() yx = seckey_x.get_shared_key(pubkey_Y, hash_nil) bx = seckey_x.get_shared_key(pubkey_B, hash_nil) # secret_input = EXP(Y,x) | EXP(B,x) | ID | B | X | Y | PROTOID secret_input = (yx + bx + node_id + pubkey_B.serialize() + pubkey_X.serialize() + pubkey_Y.serialize() + PROTOID) verify = H_verify(secret_input) # auth_input = verify | ID | B | Y | X | PROTOID | "Server" auth_input = (verify + node_id + pubkey_B.serialize() + pubkey_Y.serialize() + pubkey_X.serialize() + PROTOID + b"Server") my_auth = H_mac(auth_input) badness = my_auth != their_auth badness |= bad_result(yx) + bad_result(bx) if badness: return None return kdf_ntor(secret_input, keyBytes) # ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ def demo(node_id=b"iToldYouAboutStairs.", server_key=PrivateKey()): """ Try to handshake with ourself. """ x, create = client_part1(node_id, server_key.get_public()) skeys, created = server(server_key, node_id, create) ckeys = client_part2(x, created, node_id, server_key.get_public()) assert len(skeys) == 72 assert len(ckeys) == 72 assert skeys == ckeys print("OK") # ====================================================================== def timing(): """ Use Python's timeit module to see how fast this nonsense is """ import timeit t = timeit.Timer(stmt="ntor_ref.demo(N,SK)", setup="import ntor_ref,curve25519;N='ABCD'*5;SK=ntor_ref.PrivateKey()") print(t.timeit(number=1000)) # ====================================================================== def kdf_vectors(): """ Generate some vectors to check our KDF. """ import binascii def kdf_vec(inp): k = kdf_rfc5869(inp, T_KEY, M_EXPAND, 100) print(repr(inp), "\n\""+ binascii.b2a_hex(k)+ "\"") kdf_vec("") kdf_vec("Tor") kdf_vec("AN ALARMING ITEM TO FIND ON YOUR CREDIT-RATING STATEMENT") # ====================================================================== def test_tor(): """ Call the test-ntor-cl command-line program to make sure we can interoperate with Tor's ntor program """ enhex=lambda s: binascii.b2a_hex(s) dehex=lambda s: binascii.a2b_hex(s.strip()) PROG = b"./src/test/test-ntor-cl" def tor_client1(node_id, pubkey_B): " returns (msg, state) " p = subprocess.Popen([PROG, b"client1", enhex(node_id), enhex(pubkey_B.serialize())], stdout=subprocess.PIPE) return map(dehex, p.stdout.readlines()) def tor_server1(seckey_b, node_id, msg, n): " returns (msg, keys) " p = subprocess.Popen([PROG, "server1", enhex(seckey_b.serialize()), enhex(node_id), enhex(msg), str(n)], stdout=subprocess.PIPE) return map(dehex, p.stdout.readlines()) def tor_client2(state, msg, n): " returns (keys,) " p = subprocess.Popen([PROG, "client2", enhex(state), enhex(msg), str(n)], stdout=subprocess.PIPE) return map(dehex, p.stdout.readlines()) node_id = b"thisisatornodeid$#%^" seckey_b = PrivateKey() pubkey_B = seckey_b.get_public() # Do a pure-Tor handshake c2s_msg, c_state = tor_client1(node_id, pubkey_B) s2c_msg, s_keys = tor_server1(seckey_b, node_id, c2s_msg, 90) c_keys, = tor_client2(c_state, s2c_msg, 90) assert c_keys == s_keys assert len(c_keys) == 90 # Try a mixed handshake with Tor as the client c2s_msg, c_state = tor_client1(node_id, pubkey_B) s_keys, s2c_msg = server(seckey_b, node_id, c2s_msg, 90) c_keys, = tor_client2(c_state, s2c_msg, 90) assert c_keys == s_keys assert len(c_keys) == 90 # Now do a mixed handshake with Tor as the server c_x, c2s_msg = client_part1(node_id, pubkey_B) s2c_msg, s_keys = tor_server1(seckey_b, node_id, c2s_msg, 90) c_keys = client_part2(c_x, s2c_msg, node_id, pubkey_B, 90) assert c_keys == s_keys assert len(c_keys) == 90 print("OK") # ====================================================================== if __name__ == '__main__': if len(sys.argv) < 2: print(__doc__) elif sys.argv[1] == 'gen_kdf_vectors': kdf_vectors() elif sys.argv[1] == 'timing': timing() elif sys.argv[1] == 'self-test': demo() elif sys.argv[1] == 'test-tor': test_tor() else: print(__doc__)