From 98c12749b21417ef6a21ce6cfb703bce0a9f4f76 Mon Sep 17 00:00:00 2001 From: Gilles Peskine Date: Tue, 20 Jan 2026 14:09:30 +0100 Subject: [PATCH] New script to cross-validate SSL client and server logs Signed-off-by: Gilles Peskine --- scripts/mbedtls_framework/ssl_log_parser.py | 80 ++++++++++++++++++ scripts/validate_ssl_logs.py | 93 +++++++++++++++++++++ 2 files changed, 173 insertions(+) create mode 100644 scripts/mbedtls_framework/ssl_log_parser.py create mode 100755 scripts/validate_ssl_logs.py diff --git a/scripts/mbedtls_framework/ssl_log_parser.py b/scripts/mbedtls_framework/ssl_log_parser.py new file mode 100644 index 000000000..d62e0a756 --- /dev/null +++ b/scripts/mbedtls_framework/ssl_log_parser.py @@ -0,0 +1,80 @@ +"""Parse logs from ssl_client2 and ssl_server2.""" + +# Copyright The Mbed TLS Contributors +# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later + +import re +from typing import Dict, Iterator, List, Tuple + + +class Info: + """Information gathered from a log of ssl_client2 or ssl_server2.""" + + DUMPING_RE = re.compile(' +'.join([ + r'(?P\S+):(?P[0-9]+):', + r'\|(?P[0-9]+)\|(?: (?P
\w+):)?', + r'dumping \'(?P.*?)\'', + r'\((?P[0-9]+) bytes\)', + ])) + DUMP_CHUNK_RE = re.compile(' +'.join([ + r'(?P\S+):(?P[0-9]+):', + r'\|(?P[0-9]+)\|(?: (?P
\w+):)?', + r'[0-9a-f]+:', + r'(?P(?:[0-9a-f]{2} *){1,16})', + ])) + + def __init__(self) -> None: + """Create an empty log info object.""" + self.dumps = {} #type: Dict[str, List[str]] + + def add_dump(self, name: str, hex_data: str) -> None: + """Add a hex dump.""" + self.dumps.setdefault(name, []).append(hex_data) + + def read_dump(self, + filename: str, + lines: Iterator[Tuple[int, str]], + length: int) -> str: + """Read a hex dump. Return the hex data. + + This method consumes the data dump lines from lines. + """ + acc = '' + remaining = length + while remaining > 0: + lineno, line = next(lines) + m = self.DUMP_CHUNK_RE.match(line) + if not m: + raise Exception(f'{filename}:{lineno}: not a dump chunk as expected') + acc += m.group('data') + remaining -= 16 + plain = acc.replace(' ', '') + assert len(plain) == length * 2 + return plain + + def read_file_contents(self, + filename: str, + lines: Iterator[Tuple[int, str]]) -> None: + """Read lines from a log file and store the information we find. + + The iterator lines delivers a stream of lines with their line numbers. + """ + for _lineno, line in lines: + m = self.DUMPING_RE.match(line) + if m: + what = m.group('what') + hex_data = self.read_dump(filename, lines, + int(m.group('length'))) + self.add_dump(what, hex_data) + + def read_file(self, filename: str) -> None: + """Read a log file and store the information we find.""" + with open(filename) as input_: + self.read_file_contents(filename, enumerate(input_, 1)) + + +def parse_log_file(filename: str) -> Info: + """Parse a log of ssl_client2 or ssl_server2.""" + info = Info() + info.read_file(filename) + return info diff --git a/scripts/validate_ssl_logs.py b/scripts/validate_ssl_logs.py new file mode 100755 index 000000000..5f689c14d --- /dev/null +++ b/scripts/validate_ssl_logs.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +"""Validate logs from ssl_client2 and ssl_server2. + +On success, print nothing and return 0. +On a validation failure, print a short error message and return 1. +On a command line or parse error, die with an exception and return 1. +""" + +# Copyright The Mbed TLS Contributors +# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later + +import argparse +import sys +from typing import Callable, Dict, List, Optional + +from mbedtls_framework import ssl_log_parser + + + +def match_random(client_log: ssl_log_parser.Info, + server_log: ssl_log_parser.Info) -> Optional[str]: + """Check that both sides have the same idea of client_random and server_random.""" + client_client_randoms = client_log.dumps['client hello, random bytes'] + client_server_randoms = client_log.dumps['server hello, random bytes'] + server_client_randoms = server_log.dumps['client hello, random bytes'] + server_server_randoms = server_log.dumps['server hello, random bytes'] + if len(client_client_randoms) != len(server_client_randoms): + return ('Client and server disagree on the number of client_random ' + + f'({len(client_client_randoms)} != {len(server_client_randoms)})') + if len(client_server_randoms) != len(server_server_randoms): + return ('Client and server disagree on the number of server_random ' + + f'({len(client_server_randoms)} != {len(server_server_randoms)})') + for n, (c, s) in enumerate(zip(client_client_randoms, server_client_randoms)): + if c != s: + return f'Client and server disagree on client random #{n}' + for n, (c, s) in enumerate(zip(client_server_randoms, server_server_randoms)): + if c != s: + return f'Client and server disagree on server random #{n}' + return None + + +Task = Callable[[ssl_log_parser.Info, ssl_log_parser.Info], Optional[str]] + +TASKS = { + 'match_random': match_random, +} #type: Dict[str, Task] + +def validate(client_log: ssl_log_parser.Info, + server_log: ssl_log_parser.Info, + tasks: List[str]) -> Optional[str]: + """Perform validation tasks on a pair of matching logs. + + Return None if the validation succeeds, a human-oriented error message + otherwise. + """ + for task_name in tasks: + task = TASKS[task_name] + outcome = task(client_log, server_log) + if outcome is not None: + return outcome + return None + + +def main() -> int: + """Command line entry point.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--list-tasks', + help='List available tasks and exit') + parser.add_argument('client_log', metavar='FILE', + help='Client log file ($CLI_OUT or ?-cli-*.log)') + parser.add_argument('server_log', metavar='FILE', + help='Server log file ($SRV_OUT or ?-srv-*.log)') + parser.add_argument('tasks', metavar='TASK', + nargs='+', #action='append', + help='Tasks to perform (use --list-tasks to see supported task names)') + args = parser.parse_args() + if args.list_tasks: + for task_name in sorted(TASKS.keys()): + print(task_name) + return 0 + client_log = ssl_log_parser.parse_log_file(args.client_log) + server_log = ssl_log_parser.parse_log_file(args.server_log) + outcome = validate(client_log, server_log, args.tasks) + if outcome is None: + return 0 + else: + if outcome and outcome[-1] != '\n': + outcome += '\n' + sys.stderr.write(outcome) + return 1 + +if __name__ == '__main__': + sys.exit(main())