Commander1024
ac80863d80
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.
231 lines
12 KiB
Python
231 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Focused policy test: actual project helper plus installed IDF argv parser.
|
|
|
|
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
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
IDF = Path(os.environ.get("IDF_PATH", str(Path.home() / ".platformio/packages/framework-espidf")))
|
|
source = (ROOT / "src/admin_ssh_console.c").read_text()
|
|
start = source.index("bool admin_ssh_console_web_user_command_allowed(")
|
|
policy = source[start:source.index("\n}", start) + 2]
|
|
start = source.index("static bool remote_command_allowed(")
|
|
helper = source[start:source.index("\n}", start) + 2]
|
|
prelude = r'''
|
|
#include <assert.h>
|
|
#include <stdbool.h>
|
|
#include <stddef.h>
|
|
#include <stdint.h>
|
|
#include <string.h>
|
|
#include <stdio.h>
|
|
#define ADMIN_SSH_CONSOLE_COMMAND_LINE_CAPACITY 256U
|
|
#define ADMIN_SSH_CONSOLE_MAX_ARGUMENTS 10U
|
|
#define ADMIN_CONSOLE_TRANSPORT_WEB 1U
|
|
#define USER_DATABASE_USERNAME_CAPACITY 16U
|
|
#define USER_ROLE_ADMIN 2
|
|
typedef struct { int role; size_t username_length; char username[17]; } user_principal_t;
|
|
typedef struct {
|
|
user_principal_t principal;
|
|
struct { uint8_t transport; } token;
|
|
char line[ADMIN_SSH_CONSOLE_COMMAND_LINE_CAPACITY + 1U];
|
|
} admin_request_t;
|
|
size_t esp_console_split_argv(char *, char **, size_t);
|
|
static void secure_wipe(void *p, size_t n) {
|
|
volatile unsigned char *bytes = p;
|
|
while (n--) *bytes++ = 0;
|
|
}
|
|
'''
|
|
cases = r'''
|
|
int main(void) {
|
|
const struct { const char *line; bool allowed; } cases[] = {
|
|
{"", true}, {" ", true}, {" ", true},
|
|
{"memory", true}, {"user", true}, {"user list", true},
|
|
{"user show bootstrap", true}, {"exit", true},
|
|
/* 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\"", true},
|
|
{"\"user\" \"recover\" --force", false},
|
|
};
|
|
for (size_t i = 0; i < sizeof(cases)/sizeof(cases[0]); ++i) {
|
|
admin_request_t request = {0};
|
|
strcpy(request.line, cases[i].line);
|
|
assert(remote_command_allowed(&request) == cases[i].allowed);
|
|
assert(!strcmp(request.line, cases[i].line));
|
|
}
|
|
const char *web_allowed[] = {
|
|
"", " ", "help", "memory", "exit", "user", "user status", "user list",
|
|
"user show admin", "\"user\" \"show\" \"bootstrap\"",
|
|
"web status", "web stop", "reboot", "\"reboot\"", "\"web\" \"stop\"",
|
|
"wifi status", "mdns status", "\"web\" \"status\"",
|
|
"user add other user", "user add other admin", "user password other",
|
|
"user delete other --force", "user role other user --force",
|
|
"user role other admin --force", "\"user\" \"password\" \"other\"",
|
|
"web certificate rotate --force",
|
|
" \"web\" \"certificate\" \"rotate\" \"--force\" ",
|
|
"ssh status", "ssh sessions", "ssh counters", "ssh host-key info", "ssh start",
|
|
};
|
|
const char *web_denied[] = {
|
|
"web", "web help", "web start", "web stop extra", "web counters", "web clear-counters",
|
|
"web diagnostics enable", "web diagnostics disable", "web diagnostics show", "web diagnostics clear",
|
|
"web credentials show", "web credentials rotate --force", "web certificate info",
|
|
"web certificate rotate", "web certificate rotate --force extra",
|
|
"web certificate rotate --force --force", "web certificate rotate --Force",
|
|
"web certificate rotate --forcex", "web certificate --force rotate",
|
|
"\"web\" \"certificate\" \"rotate\" \"--force extra\"",
|
|
"web reset --force", "web status extra",
|
|
"wifi", "wifi profiles", "wifi scan", "wifi start", "wifi stop", "wifi save",
|
|
"wifi load", "wifi defaults", "wifi reset", "wifi ping example.org",
|
|
"mdns", "mdns suffix test", "mdns save", "mdns load", "mdns defaults", "mdns reset",
|
|
"reboot --force", "user bootstrap", "user recover --force",
|
|
"user add other admin --generate", "user delete other",
|
|
"user role other user", "user password admin --generate",
|
|
"user password admin", "user password other --generate",
|
|
"user delete admin --force", "user role admin admin --force",
|
|
"user role admin user --force", "user add admin admin",
|
|
"\"user\" \"password\" \"admin\"", "user password other extra",
|
|
"user add other invalid", "user add other user extra",
|
|
"user delete other --force extra", "user role other admin --force extra",
|
|
"user key add admin", "user key clear admin --force",
|
|
"user key delete admin 0 --force", "user list extra", "user show admin extra",
|
|
"ssh stop", "ssh disconnect 7", "ssh host-key rotate --force", "ssh reset --force",
|
|
" \"user\" \"password\" \"admin\" \"--generate\"",
|
|
"\"web\" \"credentials\" \"show\"", "\"wifi\" \"stop\"",
|
|
"\"mdns\" \"reset\"", "\"reboot\" extra", "\"ssh\" \"stop\"",
|
|
"\"ssh\" \"host-key\" \"rotate\" --force", "\"user\" \"recover\" --force",
|
|
};
|
|
for (size_t i=0; i<sizeof(web_allowed)/sizeof(web_allowed[0]); ++i) {
|
|
admin_request_t request={.token.transport=ADMIN_CONSOLE_TRANSPORT_WEB};
|
|
request.principal = (user_principal_t){.role=USER_ROLE_ADMIN,
|
|
.username_length=5, .username="admin"};
|
|
strcpy(request.line,web_allowed[i]);
|
|
assert(remote_command_allowed(&request));
|
|
assert(!strcmp(request.line,web_allowed[i]));
|
|
}
|
|
for (size_t i=0; i<sizeof(web_denied)/sizeof(web_denied[0]); ++i) {
|
|
admin_request_t request={.token.transport=ADMIN_CONSOLE_TRANSPORT_WEB};
|
|
request.principal = (user_principal_t){.role=USER_ROLE_ADMIN,
|
|
.username_length=5, .username="admin"};
|
|
strcpy(request.line,web_denied[i]);
|
|
if (remote_command_allowed(&request)) fprintf(stderr,"Unexpected allow: %s\n",request.line);
|
|
assert(!remote_command_allowed(&request));
|
|
assert(!strcmp(request.line,web_denied[i]));
|
|
request.token.transport=0;
|
|
/* SSH retains only the global recovery dispatcher restriction. */
|
|
assert(remote_command_allowed(&request) ==
|
|
(strstr(request.line,"recover")==NULL));
|
|
}
|
|
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:
|
|
path = Path(directory)
|
|
(path / "test.c").write_text(prelude + policy + helper + cases)
|
|
subprocess.run(["cc", "-std=c11", "-Wall", "-Wextra", "-Werror",
|
|
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")
|