Commander1024/ESP32_Serial_Swiss_Army_Knife
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove Legacy Credential Bootstrap Paths

Browse Source 
Decouple user provisioning from HTTPS identity storage while retaining
compatible v1 user records and migrating TLS material to the
credential-free
v2 format. Add focused security regression coverage and update operator
documentation.
This commit is contained in:
Commander1024
2026-09-08 19:09:26 +02:00
parent 82f21d6116
commit ac80863d80
26 changed files with 1013 additions and 583 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/admin_ssh_policy/run.py
+103 -5
View File
@@ -5,6 +5,7 @@ Requires Python 3, cc and IDF_PATH (defaults to PlatformIO's installed SDK).
Does not run FreeRTOS dispatch, SSH I/O or target hardware.
"""
import os
import re
from pathlib import Path
import subprocess
import tempfile
@@ -46,10 +47,11 @@ int main(void) {
{"", true}, {" ", true}, {" ", true},
{"memory", true}, {"user", true}, {"user list", true},
{"user show bootstrap", true}, {"exit", true},
{"user bootstrap", false}, {"user bootstrap extra", false},
/* Removed verbs reach the canonical handler, not a bootstrap policy. */
{"user bootstrap", true}, {"user bootstrap extra", true},
{"user recover", false}, {"user recover --force", false},
{" user recover --force ", false},
{"\"user\" \"bootstrap\"", false},
{"\"user\" \"bootstrap\"", true},
{"\"user\" \"recover\" --force", false},
};
for (size_t i = 0; i < sizeof(cases)/sizeof(cases[0]); ++i) {
@@ -116,11 +118,11 @@ int main(void) {
assert(!remote_command_allowed(&request));
assert(!strcmp(request.line,web_denied[i]));
request.token.transport=0;
/* SSH retains only the global bootstrap/recover dispatcher restriction. */
/* SSH retains only the global recovery dispatcher restriction. */
assert(remote_command_allowed(&request) ==
(strstr(request.line,"bootstrap")==NULL && strstr(request.line,"recover")==NULL));
(strstr(request.line,"recover")==NULL));
}
puts("PASS: SSH policy unchanged; web bounded account forms, restrictions/lifecycle and quoted forms checked with actual IDF parser");
puts("PASS: UART0-only recovery, removed bootstrap policy, web bounded account forms, restrictions/lifecycle and quoted forms checked with actual IDF parser");
}
'''
with tempfile.TemporaryDirectory(prefix="admin-ssh-policy-") as directory:
@@ -130,3 +132,99 @@ with tempfile.TemporaryDirectory(prefix="admin-ssh-policy-") as directory:
str(path / "test.c"), str(IDF / "components/console/split_argv.c"),
"-o", str(path / "test")], check=True, timeout=30)
subprocess.run([str(path / "test")], check=True, timeout=10)
# Compile the actual composition-root security initialization and start gates.
# Other subsystem setup is excluded; deterministic errors model its results.
main = (ROOT / "src/main.c").read_text()
initialization = main[main.index(" web_security_load_result_t web_security_source"):
main.index(" esp_err_t web_runtime_error")]
gates = main[main.index(" if (wifi_error == ESP_OK && web_security_error"):
main.index(" if (local_ui_error == ESP_OK)")]
web_header = "\n".join(line for line in (ROOT / "src/web_security.h").read_text().splitlines()
if not line.startswith(("#include", "#pragma once")))
user_header = (ROOT / "src/user_database.h").read_text()
user_state = re.search(r"typedef enum \{[^{}]*\} user_database_load_result_t;", user_header).group()
startup = r'''
#include <assert.h>
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>
typedef int esp_err_t;
#define ESP_OK 0
#define ESP_FAIL -1
#define SSH_TRANSPORT_PORT 22
#define TAG "test"
#define ESP_LOGI(tag, ...) snprintf(last_log, sizeof(last_log), __VA_ARGS__)
#define ESP_LOGE(tag, ...) snprintf(last_error, sizeof(last_error), __VA_ARGS__)
static char last_log[256], last_error[256], tls_log[256];
static esp_err_t tls_error, db_error;
static unsigned tls_calls, db_calls, web_starts, ssh_starts;
static const char *esp_err_to_name(esp_err_t e) { (void)e; return "error"; }
static esp_err_t web_server_start(void) { ++web_starts; return ESP_OK; }
static esp_err_t ssh_transport_start(void) { ++ssh_starts; return ESP_OK; }
'''
startup += web_header + "\n" + user_state + r'''
static web_security_load_result_t tls_source;
static user_database_load_result_t db_source;
esp_err_t web_security_init(web_security_load_result_t *out) {
++tls_calls; *out=tls_source; return tls_error;
}
esp_err_t user_database_init(user_database_load_result_t *out) {
++db_calls; strcpy(tls_log,last_log); *out=db_source; return db_error;
}
static void boot(esp_err_t random_error, esp_err_t wifi_error,
esp_err_t web_runtime_error, esp_err_t ssh_security_error,
esp_err_t ssh_runtime_error) {
'''
startup += initialization + gates + r'''
}
int main(void) {
const web_security_load_result_t sources[]={WEB_SECURITY_LOAD_STORED,
WEB_SECURITY_LOAD_GENERATED_MISSING, WEB_SECURITY_LOAD_MIGRATED_V1};
const char *labels[]={"Using stored HTTPS identity", "Using newly generated HTTPS identity",
"Using migrated v1 HTTPS identity"};
for (unsigned i=0;i<3;++i) {
tls_source=sources[i];
for (unsigned empty=0;empty<2;++empty) {
db_source=empty ? USER_DATABASE_LOAD_EMPTY : USER_DATABASE_LOAD_STORED;
boot(ESP_OK,ESP_FAIL,ESP_OK,ESP_OK,ESP_OK);
assert(!strcmp(tls_log,labels[i]));
assert(!strcmp(last_log,empty ? "Using new empty user database" : "Using stored user database"));
}
}
for (unsigned failures=0;failures<64;++failures) {
tls_error=(failures&1) ? ESP_FAIL : ESP_OK;
db_error=(failures&2) ? ESP_FAIL : ESP_OK;
esp_err_t wifi=(failures&4) ? ESP_FAIL : ESP_OK;
esp_err_t web_runtime=(failures&8) ? ESP_FAIL : ESP_OK;
esp_err_t ssh_security=(failures&16) ? ESP_FAIL : ESP_OK;
esp_err_t ssh_runtime=(failures&32) ? ESP_FAIL : ESP_OK;
tls_calls=db_calls=web_starts=ssh_starts=0;
boot(ESP_OK,wifi,web_runtime,ssh_security,ssh_runtime);
assert(tls_calls==1 && db_calls==1);
assert(web_starts==(!wifi && !tls_error && !web_runtime));
assert(ssh_starts==(!wifi && !ssh_security && !ssh_runtime));
if (db_error) assert(strstr(last_error,"user recover --force"));
}
tls_calls=db_calls=web_starts=ssh_starts=0;
boot(ESP_FAIL,ESP_OK,ESP_OK,ESP_FAIL,ESP_OK);
assert(!tls_calls && db_calls==1 && !web_starts && !ssh_starts);
puts("PASS: startup init signatures/states, exact TLS source logs, 64 independent service-gate cases and RNG failure");
}
'''
(path / "startup.c").write_text(startup)
subprocess.run(["cc", "-std=c11", "-Wall", "-Wextra", "-Werror",
str(path / "startup.c"), "-o", str(path / "startup")], check=True, timeout=30)
subprocess.run([str(path / "startup")], check=True, timeout=10)
completion = (ROOT / "src/console_completion.c").read_text()
candidates = re.findall(r'^\s*"([^"\n]+)",?$', completion, re.MULTILINE)
assert not any(c.startswith(("user bootstrap", "web credentials")) for c in candidates)
for retained in ("user recover --force", "user add", "user password", "user key add",
"web certificate info", "web certificate rotate --force", "web reset --force"):
assert retained in candidates
assert "Bootstrap/recovery" not in source
assert "legacy" not in initialization
print("PASS: removed completion entries, retained account/TLS/recovery commands and no startup credential copy")