Add SSH host identity rotation controls

2026-09-13 19:58:05 +02:00
parent aa4bbc2c8c
commit 8df1d2218b
19 changed files with 626 additions and 107 deletions
@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+"""Full production SSH security with real mbedTLS and fault-injected NVS/RTOS."""
+import ast
+import re
+from pathlib import Path
+import subprocess
+import tempfile
+ROOT = Path(__file__).resolve().parents[2]
+tree = ast.parse((ROOT / 'tests/web_security/run.py').read_text())
+headers = ast.literal_eval(next(n.value for n in tree.body if isinstance(n, ast.Assign) and any(isinstance(t, ast.Name) and t.id == 'HEADERS' for t in n.targets)))
+headers['freertos/FreeRTOS.h'] += '''
+typedef int portMUX_TYPE;
+#define portMUX_INITIALIZER_UNLOCKED 0
+void enter(void);
+void leave(void);
+#define taskENTER_CRITICAL(p) do { (void)(p); enter(); } while (0)
+#define taskEXIT_CRITICAL(p) do { (void)(p); leave(); } while (0)
+'''
+headers['freertos/task.h'] = '''#pragma once
+typedef void *TaskHandle_t;
+TaskHandle_t xTaskGetCurrentTaskHandle(void);
+void vTaskDelay(unsigned);
+'''
+with tempfile.TemporaryDirectory(prefix='ssh-security-') as directory:
+    out = Path(directory)
+    for name, text in headers.items():
+        path = out / name
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(text)
+    source = (ROOT/'src/ssh_transport.c').read_text()
+    names = ('next_generation','request_running_locked','request_running','ssh_transport_start','ssh_transport_stop','ssh_transport_replace_identity','ssh_transport_replace_host_key')
+    owner = ''.join(re.search(r'^(?:static )?[^\n]+\b'+name+r'\([^;]*?\n\{.*?^\}',source,re.M|re.S).group()+'\n' for name in names)
+    (out/'owner.inc').write_text(re.search(r'^#define SSH_TRANSPORT_GENERATION_MAX .+$',source,re.M).group()+'\n'+owner)
+    subprocess.run(['cc','-std=c11','-Wall','-Wextra','-Werror','-g','-I'+str(out),'-I'+str(ROOT/'src'),str(ROOT/'tests/ssh_management/security.c'),'-lmbedcrypto','-o',str(out/'test')],check=True,timeout=30)
+    subprocess.run([str(out/'test')],check=True,timeout=30)