#!/usr/bin/env python3 """Compare original and patched pgcrypto generators using only ephemeral keys. Requires PyCryptodome, pwntools, and openssl. Run: python3 test_generated_rsa.py /path/to/original.py /path/to/patched.py All keys are freshly generated in this process. No key parameters, generated SQL, PEM data, or serialized packets are written to files or printed. There are no embedded keys, fixtures, or pre-generated private parameters. """ from contextlib import redirect_stderr, redirect_stdout import importlib.util import io import math import subprocess import sys from unittest.mock import patch from Crypto.PublicKey import RSA def require(condition, description): if not condition: raise AssertionError(description) def load(path, name): spec = importlib.util.spec_from_file_location(name, path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def rejected(validator, components, expected_text=None): try: validator(components) except ValueError as error: if expected_text: require(expected_text in str(error), 'Unexpected validation error') return True return False def check_packet(module, components): with redirect_stderr(io.StringIO()): message = module.build_message_data(components) packet = module.build_key_data(components) require(bool(message), 'Missing generated message') require(packet[0] == 0xC7 and 192 <= packet[1] <= 223, 'Unexpected secret-subkey packet header') length = ((packet[1] - 192) << 8) + packet[2] + 192 body = packet[3:] require(len(body) == length and body[0] == 4 and body[5] == 2, 'Unexpected packet metadata') pos = 6 for field in ('n', 'e', 'd', 'p', 'q', 'u'): if field == 'd': require(body[pos] == 0, 'Unexpected secret-material encoding') pos += 1 private_start = pos bits = int.from_bytes(body[pos:pos + 2], 'big') pos += 2 size = (bits + 7) // 8 value = int.from_bytes(body[pos:pos + size], 'big') pos += size require(value == components[field] and bits == value.bit_length(), 'Serialized MPI differs from generated component') require(len(body) == pos + 2, 'Unexpected packet size') require(int.from_bytes(body[pos:], 'big') == sum(body[private_start:pos]) & 0xffff, 'Invalid packet checksum') def main(): if len(sys.argv) != 3: raise SystemExit('Pass the original and patched generator paths') original = load(sys.argv[1], 'original_generator') fixed = load(sys.argv[2], 'fixed_generator') keys = [fixed.generate_rsa_keypair() for _ in range(20)] candidate = None baseline_rejections = 0 for components in keys: phi = (components['p'] - 1) * (components['q'] - 1) only_lambda = components['d'] * components['e'] % phi != 1 require(rejected(original.validate_rsa_key, components, 'mod phi(n)') == only_lambda, 'Baseline did not distinguish phi and lambda conditions') baseline_rejections += only_lambda if only_lambda: candidate = components check_packet(fixed, components) for message in (0, 1, 42, components['p'], components['q'], components['n'] - 1): encrypted = pow(message, components['e'], components['n']) require(pow(encrypted, components['d'], components['n']) == message, 'RSA round-trip failed') print('Fresh 2048-bit keys: 20/20 validated, serialized, and round-tripped') print('Baseline rejected %d/20 of the same valid keys' % baseline_rejections) # Bound the unlikely case where the initial sample has no lambda-only key. for _ in range(12): if candidate is not None: break components = fixed.generate_rsa_keypair() phi = (components['p'] - 1) * (components['q'] - 1) if components['d'] * components['e'] % phi != 1: candidate = components require(candidate is not None, 'No distinguishing key found in 32 attempts') components = candidate phi = (components['p'] - 1) * (components['q'] - 1) require(components['d'] * components['e'] % math.lcm(components['p'] - 1, components['q'] - 1) == 1, 'Generated key does not satisfy the lambda condition') phi_inverse = dict(components, d=pow(components['e'], -1, phi)) for module in (original, fixed): module.validate_rsa_key(phi_inverse) for field in ('n', 'u'): changed = dict(phi_inverse) changed[field] += 1 require(rejected(module.validate_rsa_key, changed, 'n <> p * q' if field == 'n' else 'u * p'), 'Invalid component was accepted') require(rejected(module.validate_rsa_key, dict(phi_inverse, d=phi_inverse['d'] + 1), 'd * e'), 'Invalid private exponent was accepted') require(rejected(module.validate_rsa_key, dict(phi_inverse, e=components['p'] - 1), 'gcd'), 'Non-coprime public exponent was accepted') require(rejected(module.validate_rsa_key, dict(phi_inverse, q=components['p'], n=components['p'] ** 2), 'p = q'), 'Equal prime factors were accepted') print('Phi-inverse compatibility and invalid-component checks: PASS') key = RSA.construct(tuple(components[k] for k in ('n', 'e', 'd', 'p', 'q', 'u'))) result = subprocess.run(['openssl', 'rsa', '-check', '-noout'], input=key.export_key(), capture_output=True, timeout=15) require(result.returncode == 0 and b'RSA key ok' in result.stdout, 'OpenSSL validation failed') print('OpenSSL independently validated a freshly generated key: PASS') for module, should_pass in ((original, False), (fixed, True)): stdout, stderr = io.StringIO(), io.StringIO() with patch.object(module.RSA, 'generate', return_value=key): try: with redirect_stdout(stdout), redirect_stderr(stderr): module.main() except ValueError as error: require(not should_pass and 'mod phi(n)' in str(error), 'Unexpected full-generator failure') require(not stdout.getvalue(), 'Baseline emitted unexpected SQL') else: require(should_pass, 'Baseline unexpectedly accepted this key') require('SELECT pgp_pub_decrypt_bytea(' in stdout.getvalue(), 'Patched main did not generate SQL') print('Same ephemeral lambda-only key: original main rejects; patched main passes') stdout, stderr = io.StringIO(), io.StringIO() with redirect_stdout(stdout), redirect_stderr(stderr): fixed.main() require('SELECT pgp_pub_decrypt_bytea(' in stdout.getvalue(), 'Random generation did not produce SQL') print('Patched main with unmodified random generation: PASS') print('No private parameters or generated SQL were persisted or printed') if __name__ == '__main__': main()