196 lines
11 KiB
Python
196 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Canonical account handlers + production DB transactions; deterministic IO/NVS/crypto.
|
|
|
|
Not a concurrent RTOS, cryptographic, real-NVS or target test. Run directly.
|
|
"""
|
|
from pathlib import Path
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
IDF = Path(os.environ.get("IDF_PATH", str(Path.home() / ".platformio/packages/framework-espidf")))
|
|
|
|
def function(source, name):
|
|
start = source.index(name + "(")
|
|
start = source.rfind("\n", 0, start) + 1
|
|
return source[start:source.index("\n}", start) + 2] + "\n"
|
|
|
|
def strip_includes(text):
|
|
return "\n".join(line for line in text.splitlines()
|
|
if not line.startswith(("#include", "#pragma once")))
|
|
|
|
db = (ROOT / "src/user_database.c").read_text()
|
|
console = (ROOT / "src/user_console.c").read_text()
|
|
admin = (ROOT / "src/admin_ssh_console.c").read_text()
|
|
prelude = r'''
|
|
#define _POSIX_C_SOURCE 200809L
|
|
#include <assert.h>
|
|
#include <stdbool.h>
|
|
#include <stdint.h>
|
|
#include <stddef.h>
|
|
#include <stdio.h>
|
|
#include <string.h>
|
|
typedef int esp_err_t;
|
|
enum { ESP_OK, ESP_FAIL, ESP_ERR_INVALID_ARG, ESP_ERR_INVALID_STATE,
|
|
ESP_ERR_NO_MEM, ESP_ERR_NOT_FOUND, ESP_ERR_NOT_ALLOWED,
|
|
ESP_ERR_INVALID_RESPONSE, ESP_ERR_INVALID_VERSION, ESP_ERR_TIMEOUT };
|
|
#define pdTRUE 1
|
|
static bool snapshot_busy;
|
|
static int last_wait;
|
|
typedef void *SemaphoreHandle_t;
|
|
#define portMAX_DELAY 0
|
|
#define NVS_READWRITE 1
|
|
typedef int nvs_handle_t;
|
|
static unsigned locks, writes, commits, random_calls, prompts, checks, web_revokes, ssh_revokes;
|
|
static unsigned fail_stage, revoke_prompt, revoke_check, derivation_invalidations;
|
|
static bool invalidate_during_derivation;
|
|
static bool owner_current = true, remote = true, web = true, mismatch, cancel_prompt, stale_prompt;
|
|
static int notify_error = ESP_OK;
|
|
static char revoked_name[17];
|
|
static void secure_wipe(void *p, size_t n) { memset(p, 0, n); }
|
|
static int xSemaphoreTake(void *m, int t) { (void)m; last_wait=t; if (snapshot_busy) return 0; assert(!locks++); return pdTRUE; }
|
|
static void xSemaphoreGive(void *m) { (void)m; assert(locks-- == 1); }
|
|
static const char *esp_err_to_name(int e) { (void)e; return "injected error"; }
|
|
static int nvs_open(const char *ns, int mode, int *h) {
|
|
(void)ns; (void)mode; assert(locks);
|
|
if (invalidate_during_derivation) {
|
|
assert(derivation_invalidations==1 && !owner_current);
|
|
}
|
|
*h=1; return fail_stage==1 ? ESP_FAIL : ESP_OK;
|
|
}
|
|
static int nvs_set_blob(int h, const char *key, const void *data, size_t n) {
|
|
(void)h; (void)key; (void)data; (void)n; ++writes; return fail_stage==2 ? ESP_FAIL : ESP_OK;
|
|
}
|
|
static int nvs_commit(int h) { (void)h; ++commits; return fail_stage==3 ? ESP_FAIL : ESP_OK; }
|
|
static void nvs_close(int h) { (void)h; }
|
|
static int secure_random_fill(void *p, size_t n) {
|
|
memset(p, ++random_calls, n); return fail_stage==4 ? ESP_FAIL : ESP_OK;
|
|
}
|
|
static int derive_password(const uint8_t *p, size_t n, const uint8_t *s,
|
|
uint32_t iterations, uint8_t *hash) {
|
|
(void)p; (void)n; (void)s; (void)iterations; memset(hash, 7, 32);
|
|
if (invalidate_during_derivation) {
|
|
/* Model originating browser expiry/closure after operation admission.
|
|
* This is a deterministic derivation double, not real PBKDF2/HTTPD. */
|
|
assert(locks==1 && prompts==2 && checks==2 && owner_current);
|
|
assert(!writes && !commits);
|
|
owner_current=false;
|
|
++derivation_invalidations;
|
|
}
|
|
return fail_stage==5 ? ESP_FAIL : ESP_OK;
|
|
}
|
|
#include <openssl/sha.h>
|
|
#include <openssl/ec.h>
|
|
#include <openssl/obj_mac.h>
|
|
typedef EC_GROUP *mbedtls_ecp_group;
|
|
typedef struct { EC_POINT *point; } mbedtls_ecp_point;
|
|
#define MBEDTLS_ECP_DP_SECP256R1 1
|
|
static void mbedtls_ecp_group_init(mbedtls_ecp_group *g) { *g=NULL; }
|
|
static void mbedtls_ecp_point_init(mbedtls_ecp_point *p) { p->point=NULL; }
|
|
static int mbedtls_ecp_group_load(mbedtls_ecp_group *g, int id) {
|
|
assert(id==1); *g=EC_GROUP_new_by_curve_name(NID_X9_62_prime256v1); return *g ? 0 : -1;
|
|
}
|
|
static int mbedtls_ecp_point_read_binary(mbedtls_ecp_group *g, mbedtls_ecp_point *p, const uint8_t *b, size_t n) {
|
|
p->point=EC_POINT_new(*g); return p->point && EC_POINT_oct2point(*g,p->point,b,n,NULL)==1 ? 0 : -1;
|
|
}
|
|
static int mbedtls_ecp_check_pubkey(mbedtls_ecp_group *g, mbedtls_ecp_point *p) {
|
|
return EC_POINT_is_at_infinity(*g,p->point)==0 && EC_POINT_is_on_curve(*g,p->point,NULL)==1 ? 0 : -1;
|
|
}
|
|
static void mbedtls_ecp_point_free(mbedtls_ecp_point *p) { EC_POINT_free(p->point); }
|
|
static void mbedtls_ecp_group_free(mbedtls_ecp_group *g) { EC_GROUP_free(*g); }
|
|
static int mbedtls_sha256(const uint8_t *p, size_t n, uint8_t *h, int mode) {
|
|
assert(mode==0); return SHA256(p,n,h) ? 0 : -1;
|
|
}
|
|
'''
|
|
header = strip_includes((ROOT / "src/user_database.h").read_text())
|
|
state = db[db.index("#define USER_DATABASE_SCHEMA_VERSION"):db.index("static esp_err_t initialize_dummy_verifier(")]
|
|
fakes = r'''
|
|
|
|
static stored_database_t candidate_storage;
|
|
static user_principal_t actor;
|
|
static bool admin_ssh_console_dispatch_is_remote(void) { return remote; }
|
|
static bool admin_ssh_console_dispatch_is_web(void) { return web; }
|
|
static const user_principal_t *admin_ssh_console_dispatch_principal(void) { return remote ? &actor : NULL; }
|
|
static bool admin_ssh_console_dispatch_is_current(void) {
|
|
bool current=false; ++checks;
|
|
if (checks==revoke_check) owner_current=false;
|
|
return owner_current && user_database_principal_is_current(&actor, ¤t)==ESP_OK && current;
|
|
}
|
|
static int admin_command_gate_take(void) { return ESP_OK; }
|
|
static void admin_command_gate_give(void) {}
|
|
static int console_input_read_hidden(const char *prompt, uint8_t *out, size_t cap,
|
|
size_t min, size_t max, size_t *n) {
|
|
(void)prompt; (void)min; (void)max; assert(cap>=13); ++prompts;
|
|
memcpy(out, "test-password", 13); *n=13;
|
|
if (mismatch && prompts==2) out[0]='X';
|
|
/* Simulate invalidation just after the prompt boundary returned success. */
|
|
if (prompts==revoke_prompt) owner_current=false;
|
|
if (stale_prompt && prompts==2) ++actor.auth_generation;
|
|
return cancel_prompt ? ESP_ERR_INVALID_STATE : ESP_OK;
|
|
}
|
|
static int web_serial_transport_revoke_user(const uint8_t *u, size_t n) {
|
|
++web_revokes; assert(n<sizeof(revoked_name)); memcpy(revoked_name,u,n); revoked_name[n]=0;
|
|
assert(strcmp(revoked_name,"admin")); return notify_error;
|
|
}
|
|
static int ssh_transport_revoke_user(const uint8_t *u, size_t n) {
|
|
++ssh_revokes; assert(strlen(revoked_name)==n && !memcmp(u,revoked_name,n)); return notify_error;
|
|
}
|
|
/* Forbidden paths are traps rather than alternative implementations. */
|
|
static int show_users(const char *n) { (void)n; return 0; }
|
|
static int recover_database(void) { assert(!"recovery"); return 1; }
|
|
static int bootstrap(bool g) { (void)g; assert(!"bootstrap"); return 1; }
|
|
static int add_key(const char *n) { (void)n; assert(!"key mutation"); return 1; }
|
|
static int add_key_parts(const char *n,const uint8_t *t,size_t tl,const uint8_t *b,size_t bl) {
|
|
(void)n; (void)t; (void)tl; (void)b; (void)bl; assert(!"key mutation"); return 1;
|
|
}
|
|
esp_err_t user_database_create_generated(const uint8_t *u,size_t n,user_role_t r,user_database_generated_password_t *p) {
|
|
(void)u; (void)n; (void)r; (void)p; assert(!"generated credential"); return ESP_FAIL;
|
|
}
|
|
esp_err_t user_database_generate_password(const uint8_t *u,size_t n,user_database_generated_password_t *p) {
|
|
(void)u; (void)n; (void)p; assert(!"generated credential"); return ESP_FAIL;
|
|
}
|
|
|
|
size_t esp_console_split_argv(char *, char **, size_t);
|
|
'''
|
|
db_names = ["constant_time_equal", "all_zero", "user_database_username_valid",
|
|
"user_database_password_valid", "read_ssh_string", "user_database_key_valid",
|
|
"user_role_to_string", "user_role_parse", "set_record_password", "find_user",
|
|
"find_free_user", "stored_keys_equal", "validate_database", "recount",
|
|
"next_generation", "discard_candidate", "commit_candidate_locked", "initialize_user",
|
|
"user_database_principal_is_current", "create_locked", "user_database_create",
|
|
"mutate_user_begin", "target_matches_locked", "delete_user", "set_role",
|
|
"user_database_delete", "user_database_set_role", "user_database_get_accounts",
|
|
"user_database_delete_current", "user_database_set_role_current",
|
|
"set_password", "user_database_set_password", "user_database_set_password_current",
|
|
"user_database_generate_password_value",
|
|
"user_database_get_account_keys", "add_ssh_key", "remove_ssh_key", "clear_ssh_keys",
|
|
"user_database_add_ssh_key", "user_database_remove_ssh_key", "user_database_clear_ssh_keys",
|
|
"key_target_valid", "user_database_add_ssh_key_current", "user_database_remove_ssh_key_current",
|
|
"user_database_clear_ssh_keys_current", "fill_principal", "user_database_authorize_ssh_public_key"]
|
|
console_names = ["print_usage", "revoke_user_network_sessions", "read_password",
|
|
"show_generated_password", "mutation_currentness", "add_user", "change_password",
|
|
"parse_key_index", "command_user_inner", "command_user"]
|
|
unit = prelude + header + "\n" + state + fakes
|
|
unit += "\n".join(function(db, n) for n in db_names)
|
|
unit += function(admin, "admin_ssh_console_web_user_command_allowed")
|
|
unit += "\n".join(function(console, n) for n in console_names)
|
|
account_tests = (ROOT / "tests/admin_console_boundary/accounts.c").read_text()
|
|
key_tests = (ROOT / "tests/admin_console_boundary/account_keys.c").read_text()
|
|
account_tests = account_tests.replace('int main(void)', key_tests + '\nint main(void)')
|
|
account_tests = account_tests.replace(' typed_account_tests();', ' typed_key_tests();\n typed_account_tests();')
|
|
assert ' typed_key_tests();' in account_tests
|
|
unit += account_tests
|
|
with tempfile.TemporaryDirectory(prefix="admin-accounts-") as directory:
|
|
path = Path(directory)
|
|
(path / "test.c").write_text(unit)
|
|
subprocess.run(["cc", "-std=c11", "-Wall", "-Wextra", "-Werror", "-Wno-unused-variable",
|
|
str(path / "test.c"), str(IDF / "components/console/split_argv.c"),
|
|
"-lcrypto", "-o", str(path / "test")], check=True, timeout=30)
|
|
result = subprocess.run([str(path / "test")], check=True, timeout=10, capture_output=True, text=True)
|
|
assert "test-password" not in result.stdout
|
|
assert "Generated password for" not in result.stdout
|
|
print("PASS: canonical SSH keys: Ed25519/P256 parser and authorization, malformed/off-curve/truncated inputs, zero-wait fingerprints, stale ID/generation/recreation, duplicates/capacity, sparse indices, failed persistence and CLI parity (OpenSSL-backed curve/SHA adapters)")
|
|
print("PASS: operation-admission semantics: browser invalidated in derivation double before NVS; admitted add/password transactions still commit, only target is revoked, next command rejects; persistence failure still preserves live state (not precommit cancellation or real concurrency)")
|
|
print("PASS: canonical parsed accounts + production DB transactions: nonself isolation, prompt revocation/cancel/mismatch, currentness, persistence/RNG/derive failures, final-admin invariants, self/generated/key/recovery traps; no password output")
|