ESP32_Serial_Swiss_Army_Knife/tests/ssh_phase12/run.py

#!/usr/bin/env python3
"""Compile exact SSH listener/peer/failure functions with deterministic socket doubles.
No target, network, wolfSSH handshake, or SDK build is exercised.
"""
from pathlib import Path
import re
import subprocess
import tempfile

ROOT = Path(__file__).resolve().parents[2]
source = (ROOT / 'src/ssh_transport.c').read_text()
header = (ROOT / 'src/ssh_transport.h').read_text()

def function(name):
    match = re.search(r'^static [^\n]+\b' + name + r'\([^;]*?\n\{.*?^\}', source, re.M | re.S)
    assert match, name
    return match.group() + '\n'

constants = '\n'.join(re.search(r'^#define ' + name + r' .+$', text, re.M).group()
                      for text, name in ((header, 'SSH_TRANSPORT_PORT'),
                                         (header, 'SSH_TRANSPORT_PEER_CAPACITY'),
                                         (source, 'SSH_TRANSPORT_LISTEN_BACKLOG')))
fakes = r'''
#include <assert.h>
#include <errno.h>
#include <netinet/tcp.h>
#include <stdbool.h>
#include <stdint.h>
#include <inttypes.h>
#include <stdio.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#define ESP_OK 0
#define ESP_FAIL -1
#define SSH_TRANSPORT_MAX_SESSIONS 2U
typedef int esp_err_t;
static int s_listen_fd = -1;
static unsigned stage, fail_at, closes, depth;
static bool s_running, s_cleanup_pending;
static unsigned withdrawals;
static void mdns_service_set_ssh_available(bool available) { assert(!depth && !available && s_listen_fd == -1); ++withdrawals; }
static int s_last_error, s_lock;
static uint32_t s_management_generation;
typedef struct {
    bool close_requested;
    unsigned generation, state, session_id;
    int socket_fd;
    int64_t handshake_deadline_us;
    char peer[SSH_TRANSPORT_PEER_CAPACITY];
    void *ssh;
} ssh_slot_t;
static ssh_slot_t s_slots[2];
#define taskENTER_CRITICAL(p) do { (void)(p); assert(!depth++); } while (0)
#define taskEXIT_CRITICAL(p) do { (void)(p); assert(!--depth); } while (0)
#define request_slot_close(slot, revoked) do { assert(!(revoked)); (slot)->close_requested=true; } while (0)
static bool step(void) { return ++stage == fail_at; }
static int fake_socket(int family, int type, int protocol) {
    assert(family == AF_INET6 && type == SOCK_STREAM && protocol == IPPROTO_TCP);
    return step() ? -1 : 42;
}
static int fake_setsockopt(int fd, int level, int option, const void *value, socklen_t size) {
    assert(fd == 42 && size == sizeof(int));
    if (level == SOL_SOCKET) { assert(option == SO_REUSEADDR && *(const int *)value == 1); return 0; }
    assert(stage == 1 && level == IPPROTO_IPV6 && option == IPV6_V6ONLY && *(const int *)value == 0);
    return step() ? -1 : 0;
}
static int fake_bind(int fd, const struct sockaddr *addr, socklen_t size) {
    const struct sockaddr_in6 *v6 = (const struct sockaddr_in6 *)addr;
    assert(fd == 42 && stage == 2 && size == sizeof(*v6));
    assert(v6->sin6_family == AF_INET6 && ntohs(v6->sin6_port) == SSH_TRANSPORT_PORT);
    assert(IN6_IS_ADDR_UNSPECIFIED(&v6->sin6_addr) && !v6->sin6_scope_id);
    return step() ? -1 : 0;
}
static int fake_listen(int fd, int backlog) {
    assert(fd == 42 && stage == 3 && backlog == SSH_TRANSPORT_LISTEN_BACKLOG);
    return step() ? -1 : 0;
}
static int fake_fcntl(int fd, int op, int arg) {
    assert(fd == 42);
    assert((stage == 4 && op == F_GETFL && arg == 0) ||
           (stage == 5 && op == F_SETFL && arg == (O_NONBLOCK | O_APPEND)));
    return step() ? -1 : (op == F_GETFL ? O_APPEND : 0);
}
static int fake_close(int fd) { assert(fd == 42 && !depth); ++closes; return 0; }
static int fake_shutdown(int fd, int how) { assert(fd == 42 && how == SHUT_RDWR && !depth); return 0; }
#define socket fake_socket
#define setsockopt fake_setsockopt
#define bind fake_bind
#define listen fake_listen
#define fcntl fake_fcntl
#define close fake_close
#define shutdown fake_shutdown
'''
accept_fakes = r'''
#define SSH_TRANSPORT_ACCEPT_BUDGET 4U
#define SSH_TRANSPORT_SESSION_HANDSHAKE 1
#define SSH_TRANSPORT_HANDSHAKE_TIMEOUT_SECONDS 15U
#define WS_SUCCESS 0
static void *s_context = (void *)1;
static struct { uint64_t io_failures, tcp_connections, capacity_rejections, handshake_failures; } s_counters;
static unsigned accepts;
static int accept_error;
static void add_counter(uint64_t *counter, uint64_t value) { *counter += value; }
static int fake_accept(int fd, struct sockaddr *addr, socklen_t *length) {
    assert(fd == 42 && addr && *length == sizeof(struct sockaddr_storage));
    ++accepts; errno = accept_error;
    return accept_error ? -1 : 42;
}
#define accept fake_accept
static ssh_slot_t *find_free_slot(size_t *index) { (void)index; return NULL; }
static uint32_t make_session_id(size_t index, uint32_t generation) { (void)index; return generation; }
static int64_t esp_timer_get_time(void) { return 0; }
static void *wolfSSH_new(void *ctx) { (void)ctx; assert(0); return NULL; }
static int wolfSSH_set_fd(void *ssh, int fd) { (void)ssh; (void)fd; assert(0); return 0; }
static void set_ctx(void *ssh, void *slot) { (void)ssh; (void)slot; assert(0); }
#define wolfSSH_SetIOReadCtx set_ctx
#define wolfSSH_SetUserAuthCtx set_ctx
#define wolfSSH_SetUserAuthResultCtx set_ctx
#define wolfSSH_SetChannelReqCtx set_ctx
static bool cleanup_slot(ssh_slot_t *slot) { (void)slot; assert(0); return false; }
static void publish_slot(ssh_slot_t *slot, size_t index) { (void)slot; (void)index; assert(0); }
'''
tests = r'''
int main(void) {
    for (fail_at = 1; fail_at <= 6; ++fail_at) {
        stage = closes = 0;
        assert(create_listener() == ESP_FAIL && s_listen_fd == -1);
        assert(stage == fail_at && closes == (fail_at != 1));
    }
    fail_at = stage = closes = 0;
    assert(create_listener() == ESP_OK && s_listen_fd == 42 && stage == 6 && !closes);
    s_running = true; s_management_generation = 7;
    listener_failed();
    assert(s_listen_fd == -1 && closes == 1 && !s_running && s_cleanup_pending && withdrawals == 1);
    assert(s_last_error == ESP_FAIL && s_management_generation == 8);
    assert(s_slots[0].close_requested && s_slots[1].close_requested);
    s_management_generation = UINT32_MAX;
    listener_failed(); assert(s_management_generation == UINT32_MAX && closes == 1);

    const int retry_errors[] = { EAGAIN, EWOULDBLOCK, EINTR, ENOMEM, ENOBUFS, ENFILE, ECONNABORTED };
    for (unsigned i = 0; i < sizeof(retry_errors)/sizeof(retry_errors[0]); ++i) {
        s_listen_fd = 42; s_running = true; accepts = 0; accept_error = retry_errors[i];
        unsigned before = withdrawals;
        accept_connections();
        assert(accepts == 1 && s_listen_fd == 42 && s_running && withdrawals == before);
    }
    const int fatal_errors[] = { EBADF, EINVAL, ENOTSOCK, EOPNOTSUPP };
    for (unsigned i = 0; i < sizeof(fatal_errors)/sizeof(fatal_errors[0]); ++i) {
        s_listen_fd = 42; s_running = true; accepts = 0; accept_error = fatal_errors[i];
        unsigned before = withdrawals;
        accept_connections();
        assert(accepts == 1 && s_listen_fd == -1 && !s_running && withdrawals == before + 1);
    }
    accepts = 0; accept_connections(); assert(!accepts);
    s_listen_fd = 42; accepts = closes = 0; accept_error = 0;
    accept_connections();
    assert(accepts == SSH_TRANSPORT_ACCEPT_BUDGET && closes == accepts);
    assert(s_counters.capacity_rejections == accepts && s_counters.tcp_connections == accepts);

    struct sockaddr_storage storage = {0};
    char output[SSH_TRANSPORT_PEER_CAPACITY];
    struct sockaddr_in *v4 = (struct sockaddr_in *)&storage;
    v4->sin_family = AF_INET; v4->sin_port = htons(65535);
    assert(inet_pton(AF_INET, "192.0.2.1", &v4->sin_addr) == 1);
    format_peer(&storage, output, sizeof(output)); assert(!strcmp(output, "192.0.2.1:65535"));
    memset(&storage, 0, sizeof(storage));
    struct sockaddr_in6 *v6 = (struct sockaddr_in6 *)&storage;
    v6->sin6_family = AF_INET6; v6->sin6_port = htons(65535);
    assert(inet_pton(AF_INET6, "fe80::1", &v6->sin6_addr) == 1);
    v6->sin6_scope_id = 3;
    format_peer(&storage, output, sizeof(output)); assert(!strcmp(output, "[fe80::1%3]:65535"));
    v6->sin6_scope_id = UINT32_MAX;
    assert(inet_pton(AF_INET6, "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff", &v6->sin6_addr) == 1);
    format_peer(&storage, output, sizeof(output));
    assert(!strcmp(output, "[ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff%4294967295]:65535"));
    v6->sin6_scope_id = 0;
    assert(inet_pton(AF_INET6, "2001:db8::1", &v6->sin6_addr) == 1);
    format_peer(&storage, output, sizeof(output)); assert(!strcmp(output, "[2001:db8::1]:65535"));
    assert(inet_pton(AF_INET6, "::ffff:192.0.2.1", &v6->sin6_addr) == 1);
    format_peer(&storage, output, sizeof(output)); assert(!strcmp(output, "[::ffff:192.0.2.1]:65535"));
    format_peer(&storage, output, 8); assert(!strcmp(output, "unknown"));
    output[0] = 'x'; format_peer(&storage, output, 0); assert(output[0] == 'x');
    format_peer(&storage, output, 1); assert(output[0] == 0);
    storage.ss_family = AF_UNSPEC;
    format_peer(&storage, output, sizeof(output)); assert(!strcmp(output, "unknown"));
    puts("PASS SSH Phase12 dual-stack socket setup, all setup failures, owner listener failure, scoped/mapped peers and bounded formatting");
}
'''
with tempfile.TemporaryDirectory(prefix='ssh-phase12-') as directory:
    out = Path(directory)
    (out / 'test.c').write_text(constants + '\n' + fakes + ''.join(function(name) for name in (
        'close_socket', 'set_nonblocking', 'create_listener', 'format_peer', 'listener_failed')) + accept_fakes + function('accept_connections') + tests)
    subprocess.run(['cc', '-std=c11', '-Wall', '-Wextra', '-Werror', str(out / 'test.c'), '-o', str(out / 'test')], check=True, timeout=30)
    subprocess.run([str(out / 'test')], check=True, timeout=10)