Commander1024
fe1e2d98b4
Allow other-account add/password and forced delete/role commands through shared dispatcher and handler policy. Keep self-target, generated-secret, key, bootstrap, and recovery workflows blocked. Revalidate currentness after password prompts and before database API admission. Document that admitted mutations may finish after disconnect, while subsequent stale operations must reject. Add policy, transaction-failure, cleanup, and targeted-revocation regressions. Record completed review, passing host tests and firmware build, with target validation and M2 acceptance still pending.
132 lines
6.7 KiB
Python
132 lines
6.7 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
|
|
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},
|
|
{"user bootstrap", false}, {"user bootstrap extra", false},
|
|
{"user recover", false}, {"user recover --force", false},
|
|
{" user recover --force ", false},
|
|
{"\"user\" \"bootstrap\"", false},
|
|
{"\"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 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 bootstrap/recover dispatcher restriction. */
|
|
assert(remote_command_allowed(&request) ==
|
|
(strstr(request.line,"bootstrap")==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");
|
|
}
|
|
'''
|
|
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)
|