#!/usr/bin/env python3 """Read-only deterministic audit of the actual Android source at one repository path. Only the authored test driver, extraction script and Android/MMKV boundary stubs are durable. Actual source copies and class files exist in a self-created temporary folder, automatically removed on exit. Nothing is written into the supplied repo. """ import hashlib import json import os from pathlib import Path import shutil import subprocess import sys import tempfile ROOT_REL = Path('app/src/main/java/com/cpt/zhct/weighting') COPIES = ( 'utils/Constants.java', 'antiescape/AlarmType.java', 'antiescape/UnlockCodePolicy.java', 'antiescape/TrayRemovalConfirmationCountConfig.java', 'antiescape/TrayRemovalConfirmationCountPolicy.java', ) METHODS = ( ('HANDLE_YOUKATE_SCAN_CODE', 'activity/HomeActivity.java', 'private void handleYoukateScanCode(String scanCode)'), ('HANDLE_EMPTY_SCAN_CODE', 'activity/HomeActivity.java', 'private void handleEmptyScanCode()'), ('CLEAN_SCAN_CODE', 'hardware/YoukateHardwareManager.java', 'private String cleanScanCode(String code)'), ) def digest(data): return hashlib.sha256(data).hexdigest() def extract_method(source, signature): """Require a unique exact signature; balance braces outside literals/comments.""" if source.count(signature) != 1: raise ValueError(f'Expected one exact signature: {signature}') start = source.index(signature) pos = source.index('{', start + len(signature)) depth = 0 state = 'code' while pos < len(source): current = source[pos] following = source[pos + 1] if pos + 1 < len(source) else '' if state == 'line-comment': if current == '\n': state = 'code' elif state == 'block-comment': if current == '*' and following == '/': state = 'code' pos += 1 elif state in ('string', 'char'): if current == '\\': pos += 1 elif current == ('"' if state == 'string' else "'"): state = 'code' elif current == '/' and following == '/': state = 'line-comment' pos += 1 elif current == '/' and following == '*': state = 'block-comment' pos += 1 elif current == '"': state = 'string' elif current == "'": state = 'char' elif current == '{': depth += 1 elif current == '}': depth -= 1 if depth == 0: return source[start:pos + 1] pos += 1 raise ValueError(f'Unclosed method: {signature}') def run(command, cwd, timeout=60): result = subprocess.run(command, cwd=cwd, text=True, capture_output=True, timeout=timeout, check=False) if result.returncode: raise RuntimeError(f'Command failed ({result.returncode}): {command[0]}\n' f'{result.stdout}{result.stderr}') return result def java_tools(): candidates = [] # Select an already-installed JDK only. This never installs or downloads tools. if sys.platform == 'darwin' and Path('/usr/libexec/java_home').is_file(): for version in ('17', '1.8'): probe = subprocess.run(['/usr/libexec/java_home', '-v', version], text=True, capture_output=True, check=False) if probe.returncode == 0: candidates.append(Path(probe.stdout.strip()) / 'bin') if os.environ.get('JAVA_HOME'): candidates.append(Path(os.environ['JAVA_HOME']) / 'bin') found = shutil.which('javac') if found: candidates.append(Path(found).resolve().parent) for directory in candidates: if (directory / 'javac').is_file() and (directory / 'java').is_file(): return str(directory / 'javac'), str(directory / 'java') raise RuntimeError('An installed JDK with javac and java is required; none found.') def main(): if len(sys.argv) != 2: raise SystemExit('Usage: python3 run_audit.py /absolute/path/to/ZhctWeightingTableYoukate') repo = Path(sys.argv[1]).expanduser().resolve(strict=True) if not repo.is_dir(): raise ValueError('Repository path must be a directory') git_root = Path(run(['git', 'rev-parse', '--show-toplevel'], repo).stdout.strip()).resolve() if git_root != repo: raise ValueError('Input must be the repository root, not a nested folder') head_before = run(['git', 'rev-parse', 'HEAD'], repo).stdout.strip() status_before = run(['git', 'status', '--porcelain'], repo).stdout assets = Path(__file__).resolve().parent source_base = repo / ROOT_REL required = sorted(set(COPIES) | {entry[1] for entry in METHODS}) source_bytes = {rel: (source_base / rel).read_bytes() for rel in required} source_text = {rel: data.decode('utf-8') for rel, data in source_bytes.items()} hashes = {str(ROOT_REL / rel): digest(data) for rel, data in source_bytes.items()} template = (assets / 'flow-template.java.txt').read_text(encoding='utf-8') method_hashes = {} for marker, rel, signature in METHODS: token = '@@' + marker + '@@' if template.count(token) != 1: raise ValueError(f'Template marker must be unique: {token}') actual = extract_method(source_text[rel], signature) method_hashes[signature] = digest(actual.encode('utf-8')) template = template.replace(token, actual) if '@@' in template: raise ValueError('Unresolved template marker') stubs = json.loads((assets / 'boundary-stubs.json').read_text(encoding='utf-8')) javac, java = java_tools() # Cleanup is restricted to this directory, which this invocation creates. with tempfile.TemporaryDirectory(prefix='scan-app-source-audit-') as temporary: work = Path(temporary) sources = work / 'src' sources.mkdir() for rel, data in stubs.items(): relative = Path(rel) if relative.is_absolute() or '..' in relative.parts: raise ValueError('Unsafe stub path') output = sources / relative output.parent.mkdir(parents=True, exist_ok=True) output.write_text(data, encoding='utf-8') for rel in COPIES: output = sources / 'com/cpt/zhct/weighting' / rel output.parent.mkdir(parents=True, exist_ok=True) output.write_bytes(source_bytes[rel]) (sources / 'ExtractedActualFlow.java').write_text(template, encoding='utf-8') classes = work / 'classes' classes.mkdir() compilation = run([javac, '-encoding', 'UTF-8', '-d', str(classes)] + [str(path) for path in sorted(sources.rglob('*.java'))], work) execution = run([java, '-cp', str(classes), 'ExtractedActualFlow'], work) results = [line for line in execution.stdout.splitlines() if line.strip()] if len(results) != 17 or any(not line.startswith('PASS ') for line in results): raise AssertionError(f'Expected 17 PASS results, observed: {results}') unchanged = all((source_base / rel).read_bytes() == data for rel, data in source_bytes.items()) head_after = run(['git', 'rev-parse', 'HEAD'], repo).stdout.strip() status_after = run(['git', 'status', '--porcelain'], repo).stdout if not unchanged or head_before != head_after or status_before != status_after: raise RuntimeError('Repository changed during audit; rerun on a stable snapshot') print(json.dumps({ 'audit': 'actual-source deterministic scan-interruption audit', 'repository': str(repo), 'head': head_before, 'working_tree_clean': not status_before, 'repository_unchanged_during_audit': True, 'source_sha256': hashes, 'extracted_method_sha256': method_hashes, 'test_assets_sha256': {name: digest((assets / name).read_bytes()) for name in ('run_audit.py', 'flow-template.java.txt', 'boundary-stubs.json')}, 'javac': javac, 'result_count': len(results), 'results': results, 'boundary': ('No native SDK, Android lifecycle, UI transactions, database, ' 'network, serial port or hardware executed. PASS reproduces ' 'current behavior; it does not assert product correctness. ' 'Business source copies and class files were temporary and removed.'), }, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()