Files
ESP32_Serial_Swiss_Army_Knife/tests/admin_ssh_policy/run.py
T
Commander1024 326119812f Extend browser admin lifecycle actions
Support browser reboot and HTTPS stop through deferred control, plus
exact
`web certificate rotate --force` handoff to the dispatcher. Add typed
request
validation and focused boundary and lifecycle coverage.
2026-09-07 09:36:38 +02:00

113 lines
5.6 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("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
typedef struct {
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\"",
"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 --force",
"user role other user --force", "user password admin --generate",
"user password other", "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};
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};
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 read-only exceptions, mutations/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 + 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)