diff options
Diffstat (limited to 'controller')
| -rw-r--r-- | controller/fw/Makefile | 6 | ||||
| m--------- | controller/fw/crypto-algorithms | 0 | ||||
| -rw-r--r-- | controller/fw/src/crypto.c | 80 | ||||
| -rw-r--r-- | controller/fw/src/crypto.h | 20 | ||||
| -rw-r--r-- | controller/fw/tools/crypto_test.c | 2 | ||||
| -rw-r--r-- | controller/fw/tools/crypto_test_runner.py | 46 | ||||
| -rw-r--r-- | controller/fw/tools/presig_gen.py | 184 |
7 files changed, 205 insertions, 133 deletions
diff --git a/controller/fw/Makefile b/controller/fw/Makefile index c678963..85ab02f 100644 --- a/controller/fw/Makefile +++ b/controller/fw/Makefile @@ -62,7 +62,7 @@ PAYLOAD_DATA_BIT ?= 64 TRANSMISSION_SYMBOLS ?= 32 PRESIG_STORE_SIZE ?= 3 -PRESIG_KEYFILE ?= presig_test_key.private +PRESIG_KEYFILE ?= presig_test_key.secret PRESIG_DBFILE ?= presig_test_db.sqlite3 CC := $(PREFIX)gcc @@ -161,11 +161,11 @@ $(BUILDDIR)/generated/dsss_butter_filter.h: | $(BUILDDIR)/generated .PRECIOUS: $(BUILDDIR)/generated/crypto_presig_data.c $(BUILDDIR)/generated/crypto_presig_data.c: $(PRESIG_KEYFILE) tools/presig_gen.py | $(BUILDDIR)/generated - $(PYTHON3) tools/presig_gen.py $(PRESIG_KEYFILE) $(PRESIG_DBFILE) > $@ + $(PYTHON3) tools/presig_gen.py $(PRESIG_KEYFILE) prekey > $@ .PRECIOUS: $(PRESIG_KEYFILE) $(PRESIG_KEYFILE): - $(PYTHON3) tools/presig_gen.py -g $@ + $(PYTHON3) tools/presig_gen.py $@ keygen $(BUILDDIR)/generated: ; mkdir -p $@ diff --git a/controller/fw/crypto-algorithms b/controller/fw/crypto-algorithms new file mode 160000 +Subproject cfbde48414baacf51fc7c74f275190881f037d3 diff --git a/controller/fw/src/crypto.c b/controller/fw/src/crypto.c index 73ad783..db35745 100644 --- a/controller/fw/src/crypto.c +++ b/controller/fw/src/crypto.c @@ -1,16 +1,22 @@ +#include <assert.h> #include <unistd.h> #include <stdbool.h> #include <stdlib.h> #include <string.h> -#include <aes.h> +#include <sodium.h> #include "crypto.h" #include "simulation.h" -void debug_hexdump(const char *name, uint8_t *buf, size_t len); -void debug_hexdump(const char *name, uint8_t *buf, size_t len) { + +void debug_hexdump(const char *name, const uint8_t *buf, size_t len); +int verify_trigger_dom(const uint8_t inkey[PRESIG_MSG_LEN], + const char *domain_string, const uint8_t refkey[PRESIG_MSG_LEN]); + + +void debug_hexdump(const char *name, const uint8_t *buf, size_t len) { DEBUG_PRINTN("%20s: ", name); for (size_t i=0; i<len;) { for (size_t j=0; j<8 && i<len; i++, j++) @@ -20,29 +26,55 @@ void debug_hexdump(const char *name, uint8_t *buf, size_t len) { DEBUG_PRINTN("\n"); } -int oob_message_received(uint8_t msg[static OOB_TRIGGER_LEN]) { - struct AES_ctx ctx; - uint8_t buf[crypto_sign_BYTES]; - - for (size_t serial=0; serial<PRESIG_STORE_SIZE; serial++) { - for (size_t dom=0; dom<_TRIGGER_DOMAIN_COUNT; dom++) { - - DEBUG_PRINT("Trying domain %zd serial %zd", dom, serial); - debug_hexdump("oob_presig_iv", oob_presig_iv, sizeof(oob_presig_iv)); - - memcpy(buf, presig_store[dom][serial], crypto_sign_BYTES); - debug_hexdump("presig", buf, sizeof(buf)); - AES_init_ctx_iv(&ctx, msg, oob_presig_iv); - AES_CBC_decrypt_buffer(&ctx, buf, crypto_sign_BYTES); - debug_hexdump("decrypted", buf, sizeof(buf)); - - if (!crypto_sign_verify_detached(buf, presig_messages[dom][serial], PRESIG_MSG_LEN, oob_trigger_pubkey)) { - oob_trigger_activated(dom, presig_first_serial + serial); - return 1; - } - DEBUG_PRINTN("\n"); +/* Returns 1 for correct trigger */ +int verify_trigger_dom(const uint8_t inkey[PRESIG_MSG_LEN], + const char *domain_string, const uint8_t refkey[PRESIG_MSG_LEN]) { + uint8_t key[crypto_auth_hmacsha512_KEYBYTES]; + uint8_t key_out[crypto_auth_hmacsha512_BYTES]; + + static_assert(PRESIG_MSG_LEN <= crypto_auth_hmacsha512_KEYBYTES); + memcpy(key, inkey, PRESIG_MSG_LEN); + memset(key + PRESIG_MSG_LEN, 0, sizeof(key) - PRESIG_MSG_LEN); + DEBUG_PRINT("ds \"%s\"", domain_string); + debug_hexdump("ref", refkey, PRESIG_MSG_LEN); + + for (int i=0; i<presig_height; i++) { + DEBUG_PRINT("Verifying height rel %d abs %d", i, presig_height-i); + debug_hexdump("key", key, sizeof(key)); + (void)crypto_auth_hmacsha512(key_out, (uint8_t *)domain_string, strlen(domain_string), key); + debug_hexdump("out", key_out, sizeof(key_out)); + memcpy(key, key_out, PRESIG_MSG_LEN); + memset(key + PRESIG_MSG_LEN, 0, sizeof(key) - PRESIG_MSG_LEN); + + if (!memcmp(key, refkey, PRESIG_MSG_LEN)) + return presig_height-i; + } + + return 0; +} + +int verify_trigger(const uint8_t inkey[PRESIG_MSG_LEN], int *height_out, int *domain_out) { + int res; + for (int i=0; i<_TRIGGER_DOMAIN_COUNT; i++) { + DEBUG_PRINT("Verifying domain %d", i); + if ((res = verify_trigger_dom(inkey, presig_domain_strings[i], presig_keys[i]))) { + DEBUG_PRINT("Match!"); + if (height_out) + *height_out = res - 1; + if (domain_out) + *domain_out = i; + return 1; } } + return 0; +} + +int oob_message_received(uint8_t msg[static OOB_TRIGGER_LEN]) { + int height, domain; + if (verify_trigger(msg, &height, &domain)) { + oob_trigger_activated(domain, height); + return 1; + } return 0; } diff --git a/controller/fw/src/crypto.h b/controller/fw/src/crypto.h index 61a0da8..05a3c0d 100644 --- a/controller/fw/src/crypto.h +++ b/controller/fw/src/crypto.h @@ -3,12 +3,8 @@ #include <stdint.h> -#include <sodium.h> - - -#define OOB_TRIGGER_LEN 16 #define PRESIG_MSG_LEN 16 - +#define OOB_TRIGGER_LEN PRESIG_MSG_LEN enum trigger_domain { TRIGGER_DOMAIN_ALL, @@ -19,15 +15,15 @@ enum trigger_domain { _TRIGGER_DOMAIN_COUNT }; -extern uint8_t presig_store[_TRIGGER_DOMAIN_COUNT][PRESIG_STORE_SIZE][crypto_sign_BYTES]; -extern uint8_t oob_trigger_pubkey[crypto_sign_PUBLICKEYBYTES]; -extern uint8_t presig_messages[_TRIGGER_DOMAIN_COUNT][PRESIG_STORE_SIZE][PRESIG_MSG_LEN]; -extern uint8_t oob_presig_iv[16]; -extern int presig_first_serial; - +extern const char *presig_domain_strings[_TRIGGER_DOMAIN_COUNT]; +extern uint8_t presig_keys[_TRIGGER_DOMAIN_COUNT][PRESIG_MSG_LEN]; +extern int presig_height; +extern uint8_t presig_bundle_id[16]; +extern uint64_t bundle_timestamp; -extern void oob_trigger_activated(enum trigger_domain domain, int serial); +extern void oob_trigger_activated(enum trigger_domain domain, int height); int oob_message_received(uint8_t msg[static OOB_TRIGGER_LEN]); +int verify_trigger(const uint8_t inkey[PRESIG_MSG_LEN], int *height_out, int *domain_out); #endif /* __CRYPTO_H__ */ diff --git a/controller/fw/tools/crypto_test.c b/controller/fw/tools/crypto_test.c index 8552117..410fac2 100644 --- a/controller/fw/tools/crypto_test.c +++ b/controller/fw/tools/crypto_test.c @@ -30,7 +30,7 @@ int main(int argc, char **argv) { uint8_t auth_key[16]; - for (size_t i=0; argv[1][i+0] != '\0' && argv[1][i+1] != '\0'; i+= 2) { + for (size_t i=0; argv[1][i+0] != '\0' && argv[1][i+1] != '\0' && i/2<sizeof(auth_key); i+= 2) { char buf[3] = { argv[1][i+0], argv[1][i+1], 0}; char *endptr; auth_key[i/2] = strtoul(buf, &endptr, 16); diff --git a/controller/fw/tools/crypto_test_runner.py b/controller/fw/tools/crypto_test_runner.py new file mode 100644 index 0000000..34c8b59 --- /dev/null +++ b/controller/fw/tools/crypto_test_runner.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +import subprocess +from os import path +import binascii +import re + +import presig_gen + +def do_test(domain, value, height, root_key, binary, expect_fail=False): + auth = presig_gen.gen_at_height(domain, value, height, root_key) + auth = binascii.hexlify(auth).decode() + + output = subprocess.check_output([binary, auth]) + *lines, rc_line = output.decode().splitlines() + rc = int(re.match('^rc=(\d+)$', rc_line).group(1)) + assert expect_fail == (rc == 0) + +def run_tests(root_key, max_height, binary): + for domain, value in { + 'all': 'all', + 'vendor': presig_gen.TEST_VENDOR, + 'series': presig_gen.TEST_SERIES, + 'country': presig_gen.TEST_COUNTRY, + 'region': presig_gen.TEST_REGION, + }.items(): + for height in range(max_height): + do_test(domain, value, height, root_key, binary) + do_test(domain, 'fail', height, root_key, binary, expect_fail=True) + do_test('fail', 'fail', height, root_key, binary, expect_fail=True) + do_test('', '', height, root_key, binary, expect_fail=True) + do_test(domain, value, max_height, root_key, binary, expect_fail=True) + do_test(domain, value, max_height+1, root_key, binary, expect_fail=True) + +if __name__ == '__main__': + import argparse + parser = argparse.ArgumentParser() + parser.add_argument('keyfile', help='Root key file') + parser.add_argument('max_height', type=int, default=8, nargs='?', help='Height of generated prekeys') + default_binary = path.abspath(path.join(path.dirname(__file__), '../build/tools/crypto_test')) + parser.add_argument('binary', default=default_binary, nargs='?', help='crypto_test binary to use') + args = parser.parse_args() + + with open(args.keyfile, 'r') as f: + root_key = binascii.unhexlify(f.read().strip()) + + run_tests(root_key, args.max_height, args.binary) diff --git a/controller/fw/tools/presig_gen.py b/controller/fw/tools/presig_gen.py index 2d97391..ff9dffe 100644 --- a/controller/fw/tools/presig_gen.py +++ b/controller/fw/tools/presig_gen.py @@ -4,19 +4,14 @@ import os import sys import textwrap import uuid -import hashlib +import hmac import binascii -import sqlite3 import time +from datetime import datetime -import nacl.signing -import nacl.encoding - -from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes -from cryptography.hazmat.backends import default_backend - - +LINKING_KEY_SIZE = 16 PRESIG_VERSION = '000.001' +DOMAINS = ['all', 'country', 'region', 'vendor', 'series'] def format_hex(data, indent=4, wrap=True): indent = ' '*indent @@ -28,48 +23,47 @@ def format_hex(data, indent=4, wrap=True): return f'{{\n{par}\n}}' return par -def domain_string(domain_name, value, serial): - return f'smart reset domain string v{PRESIG_VERSION}: domain:{domain_name}={value}@{serial}' +def domain_string(domain, value): + return f'smart reset domain string v{PRESIG_VERSION}: domain:{domain}={value}' -if __name__ == '__main__': - import argparse - parser = argparse.ArgumentParser() - parser.add_argument('keyfile', help='Key file to use') - parser.add_argument('presig_db', nargs='?', help='sqlite3 dbfile for generated presig authorization keys') - parser.add_argument('-g', '--generate', action='store_true', help='Generate signing keypair') - parser.add_argument('-v', '--vendor', type=str, default='Darthenschmidt Cyberei und Verschleierungstechnik GmbH', help='Vendor name for vendor domain') - parser.add_argument('-s', '--series', type=str, default='Frobnicator v0.23.7', help='Series identifier for series domain') - parser.add_argument('-r', '--region', type=str, default='Neuland', help='Region name for region domain') - parser.add_argument('-c', '--country', type=str, default='Germany', help='Country name for country domain') - parser.add_argument('-p', '--start-serial', type=int, default=0, help='First presig serial number to use') - parser.add_argument('-n', '--presig-count', type=int, default=3, help='Number of presigs to generate') - parser.add_argument('-i', '--iv', type=str, default='safety reset oob presig iv', help='IV for presig generation') - args = parser.parse_args() +def keygen_cmd(args): + if os.path.exists(args.keyfile) and not args.force: + print("Error: keyfile already exists. We won't overwrite it. Instead please remove it manually.", + file=sys.stderr) + return 1 + + root_key = os.urandom(LINKING_KEY_SIZE) - if args.generate: - if os.path.exists(args.keyfile): - print("Error: keyfile already exists. We won't overwrite it. Instead please remove it manually.", - file=sys.stderr) - sys.exit(1) - signing_key = nacl.signing.SigningKey.generate() - with open(args.keyfile, 'wb') as f: - f.write(signing_key.encode(encoder=nacl.encoding.Base64Encoder)) - f.write(b'\n') - sys.exit(0) + with open(args.keyfile, 'wb') as f: + f.write(binascii.hexlify(root_key)) + f.write(b'\n') + return 0 +def gen_at_height(domain, value, height, key): + # nanananananana BLOCKCHAIN! + + ds = domain_string(domain, value).encode('utf-8') + + for height in range(height+1): + key = hmac.digest(key, ds, 'sha512')[:LINKING_KEY_SIZE] + + return key + +def auth_cmd(args): with open(args.keyfile, 'r') as f: - signing_key = nacl.signing.SigningKey(f.read().strip(), encoder=nacl.encoding.Base64Encoder) - pubkey_bytes = signing_key.verify_key.encode(encoder=nacl.encoding.RawEncoder) - pubkey_hash = hashlib.sha512(pubkey_bytes).digest()[:16] + root_key = binascii.unhexlify(f.read().strip()) - if not args.presig_db: - print('The presig_db parameter is required.', file=sys.stderr) - sys.exit(1) + vals = [ (domain, getattr(args, domain)) for domain in DOMAINS if getattr(args, domain) is not None ] + if not vals: + vals = [('all', 'all')] + for domain, value in vals: + auth = gen_at_height(domain, value, args.height, root_key) + print(f'{domain}="{value}" @{args.height}: {binascii.hexlify(auth).decode()}') - db = sqlite3.connect(args.presig_db) - db.execute('CREATE TABLE IF NOT EXISTS presig_authkey (timestamp, pubkey_hash, bundle_id, presig_ver, domain, value, serial, authkey)') - bundle_id = uuid.uuid4().bytes +def prekey_cmd(args): + with open(args.keyfile, 'r') as f: + root_key = binascii.unhexlify(f.read().strip()) print('#include <stdint.h>') print('#include <assert.h>') @@ -77,67 +71,71 @@ if __name__ == '__main__': print('#include "crypto.h"') print() + bundle_id = uuid.uuid4().bytes print(f'/* bundle id {binascii.hexlify(bundle_id).decode()} */') print(f'uint8_t presig_bundle_id[16] = {format_hex(bundle_id)};') - print(f'int presig_first_serial = {args.start_serial};') print() - - print(f'uint8_t oob_trigger_pubkey[crypto_sign_PUBLICKEYBYTES] = {format_hex(pubkey_bytes)};') + print(f'/* generated on {datetime.now()} */') + print(f'uint64_t bundle_timestamp = {int(time.time())};') + print() + print(f'int presig_height = {args.max_height};') print() - print('uint8_t presig_messages[_TRIGGER_DOMAIN_COUNT][PRESIG_STORE_SIZE][PRESIG_MSG_LEN] = {') - device_domains = { - 'all': 'all', - 'country': args.country, - 'region': args.region, - 'vendor': args.vendor, - 'series': args.series - } - presigs = { dom: [] for dom in device_domains } - for dom, val in device_domains.items(): - print(' {') - for i in range(args.presig_count): - serial = args.start_serial + i - ds = domain_string(dom, val, serial) - ds_hash = hashlib.sha512(ds.encode()).digest()[:16] - presigs[dom].append((ds_hash, val, serial)) - print(f' {{ /* "{ds}" */') - print(format_hex(ds_hash, indent=8, wrap=False)) - print(f' }},') - print(' },') + print('const char *presig_domain_strings[_TRIGGER_DOMAIN_COUNT] = {') + for domain in DOMAINS: + ds = domain_string(domain, getattr(args, domain)) + assert '"' not in ds + print(f' [TRIGGER_DOMAIN_{domain.upper()}] = "{ds}",') print('};') print() - presig_iv = hashlib.sha512(args.iv.encode()).digest()[:16] - print(f'uint8_t oob_presig_iv[16] = {{ /* sha512("{args.iv}")[:16] */') - print(format_hex(presig_iv, wrap=False)) - print(f'}};') + print('uint8_t presig_keys[_TRIGGER_DOMAIN_COUNT][PRESIG_MSG_LEN] = {') + for domain in DOMAINS: + key = gen_at_height(domain, getattr(args, domain), args.max_height, root_key) + print(f' [TRIGGER_DOMAIN_{domain.upper()}] = {{{format_hex(key, indent=0, wrap=False)}}},') + print('};') + + print() + print('static inline void __hack_asserts_only(void) {') + print(f' static_assert(_TRIGGER_DOMAIN_COUNT == {len(DOMAINS)});') + print(f' static_assert(PRESIG_MSG_LEN == {LINKING_KEY_SIZE});') + print('}') print() - print('uint8_t presig_store[_TRIGGER_DOMAIN_COUNT][PRESIG_STORE_SIZE][crypto_sign_BYTES] = {') - for dom, hashes in presigs.items(): - print(f' {{ /* domain {dom} */') - for ds_hash, val, serial in hashes: - authkey = os.urandom(16) - cipher = Cipher(algorithms.AES(authkey), modes.CTR(presig_iv), backend=default_backend()) - enc = cipher.encryptor() - ciphertext = enc.update(ds_hash) - assert len(enc.finalize()) == 0 +TEST_VENDOR = 'Darthenschmidt Cyberei und Verschleierungstechnik GmbH' +TEST_SERIES = 'Frobnicator v0.23.7' +TEST_REGION = 'Neuland' +TEST_COUNTRY = 'Germany' + +if __name__ == '__main__': + import argparse + parser = argparse.ArgumentParser() + parser.add_argument('keyfile', help='Key file to use') - with db: - db.execute('INSERT INTO presig_authkey VALUES (?, ?, ?, ?, ?, ?, ?, ?)', - (int(time.time()*1000), pubkey_hash, binascii.hexlify(bundle_id).decode(), PRESIG_VERSION, dom, - print(format_hex(ciphertext, indent=8, wrap=False)) - print(f' }},') + subparsers = parser.add_subparsers(title='subcommands') + keygen_parser = subparsers.add_parser('keygen', help='Generate a new key') + keygen_parser.add_argument('-f', '--force', action='store_true', help='Force overwriting existing keyfile') + keygen_parser.set_defaults(func=keygen_cmd) + + auth_parser = subparsers.add_parser('auth', help='Generate one-time authentication string') + auth_parser.add_argument('height', type=int, help='Authentication string height, counting from 0 (root key)') + auth_parser.set_defaults(func=auth_cmd) + auth_parser.add_argument('-a', '--all', action='store_const', const='all', help='Vendor name for vendor domain') + auth_parser.add_argument('-v', '--vendor', type=str, nargs='?', const=test_vendor, help='Vendor name for vendor domain') + auth_parser.add_argument('-s', '--series', type=str, nargs='?', const=test_series, help='Series identifier for series domain') + auth_parser.add_argument('-r', '--region', type=str, nargs='?', const=test_region, help='Region name for region domain') + auth_parser.add_argument('-c', '--country', type=str, nargs='?', const=test_country, help='Country name for country domain') + + prekey_parser = subparsers.add_parser('prekey', help='Generate prekey data .C source code file') + prekey_parser.add_argument('-m', '--max-height', type=int, default=8, help='Height of generated prekey') + prekey_parser.add_argument('-v', '--vendor', type=str, default=test_vendor, help='Vendor name for vendor domain') + prekey_parser.add_argument('-s', '--series', type=str, default=test_series, help='Series identifier for series domain') + prekey_parser.add_argument('-r', '--region', type=str, default=test_region, help='Region name for region domain') + prekey_parser.add_argument('-c', '--country', type=str, default=test_country, help='Country name for country domain') + prekey_parser.set_defaults(func=prekey_cmd, all='all') - print(f' }},') - print(f'}};') + args = parser.parse_args() + sys.exit(args.func(args)) - print() - print('static inline void __hack_asserts_only(void) {') - print(f' static_assert(_TRIGGER_DOMAIN_COUNT == {len(presigs)});') - print(f' static_assert(PRESIG_STORE_SIZE == {args.presig_count});') - print('}') - print() |