Commander1024/ESP32_Serial_Swiss_Army_Knife
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add restricted wolfSSH ordering fix

Browse Source 
Apply hash-pinned generated edits for CVE-2025-14942 while keeping
wolfSSH 1.4.20 managed sources unchanged. Add the ABI header overlay,
provenance records, and real state-machine interoperability contracts.
This commit is contained in:
Commander1024
2026-09-16 14:04:34 +02:00
parent 4d3bb490c9
commit bea33e1c95
28 changed files with 4653 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/sdk_security_overrides/run.py
+54 -12
View File
@@ -27,6 +27,13 @@ sdk = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = sdk
SPEC.loader.exec_module(sdk)
TLS_ENTRY = next(e for e in sdk.ENTRIES if e.name == "esp_tls_mbedtls")
SOURCE_ENTRIES = tuple(e for e in sdk.ENTRIES if not e.header)
def generated_path(binary, entry):
base = binary / "security_overrides"
return (base / "wolfssh_include/wolfssh" / Path(entry.source).name if entry.header
else base / entry.name / Path(entry.source).name)
def source_path(entry, idf, project=ROOT):
@@ -91,17 +98,32 @@ def generator_tests(idf, work):
assert before == {p: (p.read_bytes(), p.stat().st_mtime_ns) for p in before}
for entry in sdk.ENTRIES:
original = source_path(entry, idf).read_bytes()
derived = (binary / "security_overrides" / entry.name / Path(entry.source).name).read_bytes()
derived = generated_path(binary, entry).read_bytes()
notice = sdk.MODIFICATION_NOTICE.encode()
assert derived.startswith(notice)
assert derived.count(notice) == 1
assert b"Modified by the ESP32_serial_swiss_army_knife project on 2026-09-15" in notice
assert derived[len(notice):].startswith(original[:original.index(b"*/") + 2])
assert original[:original.index(b"*/") + 2] in derived[:2500]
if entry.component == "wolfssl__wolfssh":
assert b"Ordering profile modified 2026-09-16" in derived[:1000]
assert derived != original
expect_error(lambda: sdk.render_entry(replace(sdk.ENTRIES[0], target="mbedtls"),
{"idf": idf, "project": ROOT}), "invalid nested target")
expect_error(lambda: sdk.render_entry(replace(sdk.ENTRIES[1], target="unknown"),
{"idf": idf, "project": ROOT}), "invalid nested target")
expect_error(lambda: sdk.render_entry(replace(sdk.ENTRIES[0], header=True),
{"idf": idf, "project": ROOT}), "unaudited header overlay")
project_copy = work / "header_mismatch"
for entry in sdk.ENTRIES:
if entry.root == "project":
copied = project_copy / entry.source
copied.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(ROOT / entry.source, copied)
header = next(e for e in sdk.ENTRIES if e.header)
(project_copy / header.source).write_bytes(b"changed header")
header_failed = work / "header_failed"
expect_error(lambda: sdk.generate(idf, project_copy, header_failed), "SHA256 mismatch")
assert not header_failed.exists(), "header drift must reject the entire source/ABI plan"
expect_error(lambda: sdk.apply_edits("x", (sdk.Edit("missing", "z"),)), "got 0")
expect_error(lambda: sdk.apply_edits("xx", (sdk.Edit("x", "z"),)), "got 2")
expect_error(lambda: sdk.generate(idf, ROOT, binary, ()), "absent")
@@ -136,7 +158,7 @@ def generator_tests(idf, work):
def extracted_tests(idf, binary, work):
texts = {e.name: (binary / "security_overrides" / e.name / Path(e.source).name).read_text() for e in sdk.ENTRIES}
texts = {e.name: generated_path(binary, e).read_text() for e in sdk.ENTRIES}
aux = {}
for rel, expected in AUXILIARY.items():
raw = (idf / rel).read_bytes()
@@ -224,6 +246,8 @@ def cmake_fixture_tests(idf, work):
copied = fixture / e.source
copied.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(source_path(e, idf), copied)
if e.header:
continue
target = e.target or f"test_{e.component}"
owner_lines = nested_lines if e.target else lines
registered_source = Path(e.source).name if e.target else source_path(e, idf, fixture)
@@ -245,11 +269,16 @@ def cmake_fixture_tests(idf, work):
'if(TEST_AMBIGUOUS)', f'set_property(TARGET test_{sdk.ENTRIES[0].component} APPEND PROPERTY SOURCES "{source_path(sdk.ENTRIES[0], idf, fixture)}")', 'endif()',
'if(TEST_TARGET_MISSING)', 'function(idf_component_get_property out component property)',
'set(${out} nonexistent PARENT_SCOPE)', 'endfunction()', 'endif()']
for e in sdk.ENTRIES:
for e in SOURCE_ENTRIES:
directory = f' DIRECTORY "{nested_dir}"' if e.target else ''
lines += [f'set_source_files_properties("{source_path(e, idf, fixture)}"{directory} PROPERTIES COMPILE_FLAGS "-DSOURCE_FLAG" COMPILE_DEFINITIONS "SOURCE_DEFINE" COMPILE_OPTIONS "-fno-common")']
lines += [f'include("{ROOT / "cmake/security_overrides.cmake"}")']
for e in sdk.ENTRIES:
(fixture / 'consumer.c').write_text('int consumer(void) { return 0; }\n')
lines += ['add_library(direct_consumer STATIC consumer.c)',
'target_link_libraries(direct_consumer PUBLIC test_wolfssl__wolfssh)',
'add_library(transitive_consumer STATIC consumer.c)',
'target_link_libraries(transitive_consumer PRIVATE direct_consumer)',
f'include("{ROOT / "cmake/security_overrides.cmake"}")']
for e in SOURCE_ENTRIES:
target = e.target or f"test_{e.component}"
directory = f' DIRECTORY "{nested_dir}"' if e.target else ''
lines += [f'file(GENERATE OUTPUT "${{CMAKE_BINARY_DIR}}/{e.name}.sources" CONTENT "$<TARGET_PROPERTY:{target},SOURCES>")',
@@ -260,24 +289,33 @@ def cmake_fixture_tests(idf, work):
(fixture / "CMakeLists.txt").write_text("\n".join(lines) + "\n")
build = work / "cmake_good"
run(["cmake", "-G", "Ninja", "-S", fixture, "-B", build])
for e in sdk.ENTRIES:
for e in SOURCE_ENTRIES:
source = (build / (e.name + ".sources")).read_text()
assert source.split(';').count(str(build / "security_overrides" / e.name / Path(e.source).name)) == 1
assert str(source_path(e, idf, fixture)) not in source
build_registration(build, idf, fixture)
commands = json.loads((build / "compile_commands.json").read_text())
for e in sdk.ENTRIES:
generated = str(build / "security_overrides" / e.name / Path(e.source).name)
for e in SOURCE_ENTRIES:
generated = str(generated_path(build, e))
matches = [c for c in commands if c["file"] == generated]
assert len(matches) == 1, (e.name, matches)
for option in ("-DSOURCE_FLAG", "-DSOURCE_DEFINE", "-fno-common",
str(source_path(e, idf, fixture).parent)):
assert option in matches[0]["command"], (e.name, option, matches)
header = next(e for e in sdk.ENTRIES if e.header)
overlay = str(generated_path(build, header))
consumers = [c for c in commands if c['file'].endswith('/consumer.c') or
'/wolfssh_internal/' in c['file'] or '/wolfssh_ssh/' in c['file']]
assert len(consumers) == 4
for command in consumers:
assert '-include' + overlay in command['command']
assert str(build / 'security_overrides/wolfssh_include') in command['command']
print('PUBLIC forced header/overlay reaches library, direct and transitive consumers PASS')
ninja = (build / "build.ninja").read_text()
for path in [ROOT / "tools/security_overrides.py", idf / "components/esp_common/include/esp_idf_version.h"] + [source_path(e, idf, fixture) for e in sdk.ENTRIES]:
for path in [ROOT / "tools/wolfssh_order/delta.json", ROOT / "tools/security_overrides.py", idf / "components/esp_common/include/esp_idf_version.h"] + [source_path(e, idf, fixture) for e in sdk.ENTRIES]:
assert str(path) in next(line for line in ninja.splitlines() if ": RERUN_CMAKE" in line), path
for flag, phrase in (("TEST_MISSING", "found 0"), ("TEST_AMBIGUOUS", "found 2"),
("TEST_TARGET_MISSING", "missing component target"),
("TEST_TARGET_MISSING", "missing wolfSSH overlay target"),
("TEST_NESTED_MISSING", "missing nested target"),
("TEST_NESTED_OWNER", "unexpected nested target owner"),
("TEST_mbedtls_MISSING_SOURCE", "found 0"),
@@ -345,7 +383,11 @@ def build_registration(build, idf, project=ROOT):
ninja = (build / "build.ninja").read_text()
compile_lines = [line for line in ninja.splitlines() if ": C_COMPILER" in line]
for e in sdk.ENTRIES:
generated = build / "security_overrides" / e.name / Path(e.source).name
generated = generated_path(build, e)
if e.header:
assert not any(str(generated) in line for line in compile_lines)
assert generated.read_bytes() == sdk.render_entry(e, {"idf": idf, "project": project})[1]
continue
matches = [line for line in compile_lines if str(generated) in line]
assert len(matches) == 1, (e.name, matches)
assert not any(str(source_path(e, idf, project)) in line for line in compile_lines), e.name
tests/ssh_protocol_policy/run.py
+33 -3
View File
@@ -117,6 +117,28 @@ def enum_containing(source, token):
return matches[0] + "\n"
def reviewed_order_function(name, original):
"""Independent, exact allowlist; do not accept an arbitrary generator delta."""
if name == 'SendExtInfo':
return ('int SendExtInfo(WOLFSSH* ssh)\n{\n'
' WOLFSSH_UNUSED(ssh);\n return WS_NOT_COMPILED;\n}\n')
if name == 'SendKexInit':
edits = (
(' ssh->isKeying = 1;',
' /* Set self is keying flag since we started sending the KEX init msg */\n'
' ssh->isKeying |= WOLFSSH_SELF_IS_KEYING;'),
(' if (ssh->ctx->side == WOLFSSH_ENDPOINT_CLIENT) {\n'
' kexAlgoNamesPlus = ",ext-info-c";\n'
' kexAlgoNamesPlusSz = (word32)WSTRLEN(kexAlgoNamesPlus);\n }\n\n', ''),
(' if (ret == WS_SUCCESS)\n ret = wolfSSH_SendPacket(ssh);',
' if (ret == WS_SUCCESS) {\n ret = wolfSSH_SendPacket(ssh);\n }'),
)
for old, new in edits:
assert original.count(old) == 1, name
original = original.replace(old, new)
return original
def main():
parser = argparse.ArgumentParser(description=__doc__)
databases = sorted((ROOT / ".pio/build").glob("*/compile_commands.json"))
@@ -182,15 +204,23 @@ def main():
functions = ("NameToId", "IdToName", "AlgoListSz", "CopyNameList",
"CopyNameListPlus", "BuildNameList", "SendKexInit", "SendExtInfo")
actual = "\n".join(extract(sources["ssh.c"], "wolfSSH_CTX_SetAlgoList" + field)
for field in FIELDS)
ssh_entry = next(e for e in ENTRIES if e.name == 'wolfssh_ssh')
_, generated_ssh = render_entry(ssh_entry, {'project': ROOT})
actual = ''
for field in FIELDS:
name = 'wolfSSH_CTX_SetAlgoList' + field
body = extract(generated_ssh.decode(), name)
assert body == extract(sources['ssh.c'], name), name
actual += body
for name in functions:
if extract(internal, name) != extract(sources["internal.c"], name):
if extract(internal, name) != reviewed_order_function(name, extract(sources["internal.c"], name)):
raise RuntimeError(f"Override changed reviewed protocol function: {name}")
actual += "\n".join(extract(internal, name) for name in functions)
# Preserve actual conditional enum values and feature-filtered name table.
types = "\n".join(enum_containing(resolved, token) for token in
("ID_NONE", "TYPE_KEX", "MSGID_KEXINIT", "WOLFSSH_ENDPOINT_SERVER"))
assert macros['WOLFSSH_SELF_IS_KEYING'] == '0x02'
types += '#define WOLFSSH_SELF_IS_KEYING 0x02\n'
types += "typedef struct { byte id; byte type; const char *name; } NameIdPair;\n" + mapping
assignments = []
for field in FIELDS:
tests/ssh_protocol_policy/vendor.c
+9 -11
View File
@@ -7,6 +7,7 @@
#include "resolved.h"
#define WLOG(...) ((void)0)
#define WOLFSSH_UNUSED(x) ((void)(x))
#define INLINE inline
#define WMEMCPY memcpy
#define WSTRLEN strlen
@@ -18,9 +19,7 @@
#define LENGTH_SZ 4U
#define BOOLEAN_SZ 1U
#define COOKIE_SZ 16U
#define WS_EXTINFO_EXTENSION_COUNT 1
static const char cannedNoneNames[] = "none";
static const char serverSigAlgsName[] = "server-sig-algs";
typedef struct { byte *kexInit; word32 kexInitSz; } HandshakeInfo;
typedef struct {
@@ -206,14 +205,13 @@ int main(void)
assert(SendKexInit(ssh) == WS_SUCCESS);
check_kex(ssh);
assert(sends == 2 && allocations == 2 && frees == 1);
assert(SendExtInfo(ssh) == WS_SUCCESS);
size_t offset = 1, length = ssh->outputBuffer.length - 8U;
const byte *p = packet + 8;
assert(p[0] == MSGID_EXT_INFO);
assert(take_u32(p, length, &offset) == 1);
expect_name(p, length, &offset, "server-sig-algs");
expect_name(p, length, &offset, "ssh-ed25519,ecdsa-sha2-nistp256");
assert(offset == length);
assert(ssh->isKeying == WOLFSSH_SELF_IS_KEYING);
assert(SendExtInfo(ssh) == WS_NOT_COMPILED);
assert(sends == 2);
ctx.side = WOLFSSH_ENDPOINT_CLIENT;
assert(SendKexInit(ssh) == WS_SUCCESS);
check_kex(ssh); /* Client must not append ext-info-c either. */
ctx.side = WOLFSSH_ENDPOINT_SERVER;
/* No key and injected packet/allocation failures must not send a fallback. */
unsigned before = sends;
ctx.privateKeyCount = 0;
@@ -234,6 +232,6 @@ int main(void)
assert(purges == old_purges);
check_kex(ssh);
bounded_free(handshake.kexInit);
puts("PASS: resolved vendor name/ID/type map, actual setters, initial/rekey KEXINIT both directions, server-sig-algs, bounded failure paths");
puts("PASS: resolved vendor name/ID/type map, actual setters, initial/rekey KEXINIT both directions, no EXT_INFO negotiation, bounded failure paths");
return 0;
}
tests/wolf_crypto_policy/run.py
+1 -1
View File
@@ -207,7 +207,7 @@ def profiles(database, candidate):
'wolfcrypt/src/curve25519.c', 'wolfcrypt/src/ed25519.c',
'wolfcrypt/src/fe_operations.c', 'wolfcrypt/src/ge_operations.c',
'wolfcrypt/src/fe_low_mem.c', 'wolfcrypt/src/ge_low_mem.c',
'wolfssl__wolfssh/src/ssh.c',
'security_overrides/wolfssh_ssh/ssh.c',
'security_overrides/wolfssh_internal/internal.c',
'src/ssh_transport.c', 'src/ssh_security.c')
for suffix in suffixes:
tests/wolfssh_order_contract/README.md
+182
View File
@@ -0,0 +1,182 @@
# Generated wolfSSH ordering contract
Implementation/provenance and restricted-profile rationale:
[`tools/wolfssh_order/README.md`](../../tools/wolfssh_order/README.md).
```sh
python3 tests/wolfssh_order_contract/run.py
python3 tests/wolfssh_order_contract/run.py --interop --target-contracts
python3 tests/wolfssh_order_contract/run.py --interop --interop-repeat 3
python3 tests/wolfssh_parser_contract/run.py
python3 tests/wolfssh_auth_contract/run.py --host-only
python3 tests/sdk_security_overrides/run.py
python3 tests/wolf_crypto_policy/run.py --host-only
```
Requires the already-installed pinned managed components, a host C compiler,
and (for `--interop`) OpenSSH `ssh`/`ssh-keygen` with `ProxyUseFdpass`, OpenSSL,
and POSIX Unix sockets with `SCM_RIGHTS` descriptor passing. The
optional `--target-contracts` also requires the existing ESP-IDF compiler and
`.pio/build/esp32-s3-devkitc-1-n16r8/compile_commands.json`. It **does not run
PlatformIO**. Everything generated by this suite is in a temporary directory;
it never writes managed sources, persistent keys or the production build tree.
No downloads or IP network sockets: a short-lived ProxyCommand passes a temporary
local Unix socket to OpenSSH. Python independently owns/reaps the server, whose
stdio uses the accepted socket. User/global SSH configuration and agents are disabled.
Sandboxes that prohibit even AF_UNIX sockets require permission for `--interop`.
All keys
and the fixed test password are disposable fixtures, not production credentials.
## Follow-up fixes and validation — 2026-09-16
`python3 tests/wolfssh_order_contract/pio_adapter.py` executes the installed
PlatformIO `get_app_flags` function and SCons ParseFlags/AppendUnique on the
configured project flags. Both forced headers must remain joined `-include/path`
arguments. A real Xtensa `-c/-o` consumer compiles; a split-option mutation must
reproduce the multiple-input failure. Also available as `run.py --pio-adapter`.
Real EOF, shutdown and exit-status tests now cover SELF, PEER and SELF|PEER,
empty/pending output, stale WANT_WRITE and repeated calls: no bytes, callbacks,
sequence changes, expectation consumption, `eofTxd` or `closeTxd` mutation.
EOF succeeds exactly once after keying clears. The misplaced PR793 EOF guard is
corrected; the additional exit-status guard is intentionally retained. Removing
the EOF guard is the seventh required failing mutation. Latest run: 8,028 checks.
Authorized firmware build PASS: `pio run`, 21.31 s, **94,340 B RAM / 1,768,701 B
flash**. Strict SDK registration, auth, protocol and crypto suites PASS against
that build. This supersedes the initial no-build/stale-artifact status below;
no device validation occurred. The 12-session OpenSSH matrix was not rerun in
this narrowly scoped follow-up; its previous evidence remains historical.
## What executes
The generator verifies original source hashes and applies the real checked-in
edits. The test includes the **entire generated `internal.c`**, links generated
`ssh.c`, unmodified pinned IO/log/port sources, and real wolfCrypt code. Static
functions are directly visible to the test; they are not rewritten copies or
models. Every test translation unit uses the generated ABI header. The original
include directory is deliberately searched first to exercise the forced overlay.
An original-header-before-overlay compile must fail explicitly.
The host crypto settings enable portable small X25519/Ed25519, TFM P-256,
import/shared-secret validation and AES-GCM. They are **host settings**, not an
assertion that every ESP compile option is identical. UBSan trap instrumentation
is enabled; the host's standalone UBSan runtime is unavailable. `WOLFSSL_USE_ALIGN`
selects bytewise encoding so x86 unaligned fast-path stores do not mask protocol
tests with alignment traps. Production crypto policy is not changed.
Coverage:
- Every byte-sized message ID at initial state for both roles; every ID during
all combinations of self/peer keying; explicit service/auth phase boundaries.
- Actual `DoPacket` rejects malformed pre-auth auth/channel/extension payloads
**before dispatch**, for both roles, without consuming the input.
- Missing, wrong, duplicate and prematurely received KEX/NEWKEYS messages;
benign transport notifications do not consume an expectation.
- Wrong optimistic INIT guess preserves the real INIT expectation; duplicate
`DoKexInit` fails before parsing another exchange.
- Actual NEWKEYS framing, queued bytes, sequence counter and key installation:
zero-through-complete one-byte write quotas, repeated output flush, no duplicate
NEWKEYS, independent self/peer bits and exactly-once handshake disposal.
Fatal IO and real invalid-AES-key errors retain the required keying state.
- Generated client/server complete password-authenticated handshakes with both
KEX algorithms and AES256-GCM, followed by server-, client- and simultaneously
initiated rekeys. IO fragments to 11/13 bytes and injects WANT_WRITE regularly.
- Six deliberately bad generated-source mutations must fail: pre-auth injection,
wrong expected KEX, unnegotiated EXT_INFO, clearing both keying bits, accepting
peer NEWKEYS before local NEWKEYS, and removing PR921's server expectation.
- Archived upstream patch hashes/commit IDs, source/header pins and unchanged
logging ABI are independently checked.
## OpenSSH interoperability (`--interop`)
Twelve real sessions cover:
- X25519 and P-256 KEX, P-256 host identity and AES128-GCM.
- Ed25519 public-key, P-256 public-key and password authentication.
- Client rekey every 32 KiB, or a server-initiated rekey with 31/37-byte fragmented
IO and forced WANT_WRITE every third send callback.
- Exact 256 KiB binary echo per session, successful channel close/status,
at least two completed exchanges and **no EXT_INFO/server-sig-algs received**.
Verified with OpenSSH **10.2p1 / OpenSSL 3.5.8** on 2026-09-16. The unfragmented
client-rekey fixtures completed ten exchanges each; fragmented server-rekey
fixtures completed two and forced roughly 3,600 write stalls each. These are
host interoperability results, not device measurements or universal-client
claims. Test-only authentication authorizes a generated key blob or a synthetic
password; public-key signature verification is still performed by real wolfSSH.
### Closure-race diagnosis and regression — 2026-09-16
The original harness failure was reproduced: exact echo and `INTEROP PASS` were
followed by OpenSSH `Broken pipe` while sending channel close. The harness called
`wolfSSH_shutdown()` once, accepted WANT_READ as retryable, drained only output,
and exited without receiving the peer's close. Flushing is not shutdown completion.
A second lifecycle issue is that OpenSSH terminates its ProxyCommand on exit;
keeping the server itself as that proxy cannot reliably prove server completion.
The harness now queues shutdown once, flushes output, pumps the real worker until
`WS_CHANNEL_CLOSED` and channel removal, then keeps receiving through transport
EOF. It accepts the expected socket-close error only with an actual zero-length
transport read, the library closed flag, and no reset. Python owns the server
independently via a local fd-passing proxy; no sleep-based grace period or return
code waiver is used. The existing 100-us polling backoff is not a close deadline.
Shutdown remains unfragmented, as before; this does not expand nonblocking-shutdown
coverage beyond the separate contract tests.
Each session requires **both process exit codes zero**, byte-for-byte 262,144-byte
echo, a single complete server evidence record, at least two completed exchanges,
the expected signed/password authentication counts, stalls in fragmented cases,
peer channel-close completion and transport EOF. OpenSSH must independently report
receipt of exit-status and channel close and `Exit status 0`; EXT_INFO remains
forbidden. Early server exit, missing evidence, nonzero client/server status,
truncated/corrupted echo, and missing rekey evidence all fail.
`--interop-repeat N` repeats the full 12-session matrix (1–20, default 1), compiling
the contracts once and generating new disposable keys per matrix. Bounds: 10 s
proxy connection, 30 s server alarm, 45 s client communication, 5 s server reap;
failed/timed-out processes are killed and reaped. No production code changes are
needed for the reproduced harness race. Final validation on OpenSSH 10.2p1:
`python3 tests/wolfssh_order_contract/run.py --interop --interop-repeat 3` passed
**36/36 sessions**, plus 8,028 contract checks and seven rejected mutations.
Client-rekey cases completed ten exchanges and server-rekey cases two; each
session had exact echo and both clean process exits. An earlier revision of the
fix also passed 36 sessions before the final close/EOF-error assertions were added.
These results do not establish general library shutdown correctness or hardware
behavior.
## Candidate target contracts (`--target-contracts`)
Reads, but does not modify, existing compiler commands. Preserves their flags
and crypto definitions while substituting freshly generated SSH sources and the
PUBLIC-equivalent header overlay in a **temporary candidate database**. Runs:
- Xtensa syntax checks for generated `internal.c`/`ssh.c` and SSH application
consumers.
- `ssh_protocol_policy`, including independently specified exact changes to
KEXINIT and disabled EXT_INFO serialization (not an arbitrary delta allowlist).
- `wolf_crypto_policy`, including real vendor vectors, independently checked
protected crypto/parser function bodies and resolved compiler flags.
- `wolfssh_auth_contract`, including password bounds/wiping/async contracts.
Those existing suites label some checks “production compiler/profile”; in this
mode they use **production-derived flags with candidate source/header paths**.
The wrapper explicitly labels the run candidate-only. This does not establish
actual firmware build registration, linking, flashing, resource reserve or
hardware behavior. Their ordinary strict commands correctly reject the old
production generated source until a firmware reconfigure is performed.
The SDK override suite separately executes the actual CMake module against a
mock IDF target graph: source ownership/properties, header hash-drift atomic
failure, PUBLIC overlay/forced include propagation to direct and transitive
consumers, configure dependencies and rejection cases. This is real CMake wiring
evidence, not a full ESP-IDF firmware build.
## Limits
No whole-upstream-fix claim. EXT_INFO and keyboard-interactive are excluded; KEX
is intentionally limited to the project algorithms. No RSA-SHA2 discovery,
DH/GEX/PQ, arbitrary-client, hardware, timing, memory-headroom, or complete SSH
parser audit claim. Existing parser/password protections remain covered by their
own suites. Firmware reconfigure/build and whole-phase device validation remain
follow-up work, deliberately not executed in this task.
tests/wolfssh_order_contract/contract.c
+288
View File
@@ -0,0 +1,288 @@
/* SPDX-License-Identifier: GPL-3.0-only */
#include <assert.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
/* The entire freshly generated translation unit, not a model of its gates. */
#include "internal.c"
static unsigned checks;
#define CHECK(x) do { ++checks; if (!(x)) { \
fprintf(stderr, "FAIL line %d: %s\n", __LINE__, #x); abort(); } } while (0)
typedef struct {
byte bytes[32768];
word32 size;
int quota, chunk, calls, fatal;
} Sink;
static int send_test(WOLFSSH* ssh, void* buf, word32 size, void* context)
{
Sink* sink = context;
(void)ssh;
++sink->calls;
if (sink->fatal) return WS_CBIO_ERR_GENERAL;
if (sink->quota == 0) return WS_CBIO_ERR_WANT_WRITE;
if (sink->quota > 0) --sink->quota;
if (size > (word32)sink->chunk) size = sink->chunk;
CHECK(size <= sizeof(sink->bytes) - sink->size);
memcpy(sink->bytes + sink->size, buf, size);
sink->size += size;
return (int)size;
}
static void policy(WOLFSSH_CTX* ctx)
{
CHECK(wolfSSH_CTX_SetAlgoListKex(ctx, "curve25519-sha256,ecdh-sha2-nistp256") == 0);
CHECK(wolfSSH_CTX_SetAlgoListKey(ctx, "ecdsa-sha2-nistp256") == 0);
CHECK(wolfSSH_CTX_SetAlgoListCipher(ctx, "aes128-gcm@openssh.com,aes256-gcm@openssh.com") == 0);
CHECK(wolfSSH_CTX_SetAlgoListMac(ctx, "hmac-sha2-256") == 0);
CHECK(wolfSSH_CTX_SetAlgoListKeyAccepted(ctx, "ssh-ed25519,ecdsa-sha2-nistp256") == 0);
}
static void gates(WOLFSSH* ssh)
{
byte consumed;
int side, state, msg;
for (side = WOLFSSH_ENDPOINT_SERVER; side <= WOLFSSH_ENDPOINT_CLIENT; ++side) {
ssh->ctx->side = side;
ssh->acceptState = ACCEPT_BEGIN;
ssh->connectState = CONNECT_BEGIN;
for (msg = 0; msg < 256; ++msg) {
ssh->isKeying = 0;
int expect = (msg >= MSGID_DISCONNECT && msg <= MSGID_DEBUG) || msg == MSGID_KEXINIT;
CHECK(IsMessageAllowed(ssh, msg, WS_MSG_RECV) == expect);
}
for (state = 1; state <= 3; ++state) {
ssh->isKeying = state;
ssh->acceptState = ACCEPT_SERVER_USERAUTH_SENT;
ssh->connectState = CONNECT_SERVER_USERAUTH_ACCEPT_DONE;
for (msg = 0; msg < 256; ++msg) {
ssh->handshake->expectMsgId = MSGID_NEWKEYS;
int result = IsMessageAllowed(ssh, msg, WS_MSG_RECV);
if (state & WOLFSSH_PEER_IS_KEYING) {
CHECK(result == ((msg >= 1 && msg <= 4) || msg == MSGID_NEWKEYS));
CHECK(ssh->handshake->expectMsgId == (msg == MSGID_NEWKEYS ? MSGID_NONE : MSGID_NEWKEYS));
}
CHECK(!IsMessageAllowed(ssh, msg, WS_MSG_SEND));
}
}
ssh->isKeying = WOLFSSH_PEER_IS_KEYING;
ssh->handshake->expectMsgId = MSGID_NONE;
CHECK(!IsMessageAllowed(ssh, MSGID_NEWKEYS, WS_MSG_RECV));
CHECK(!IsMessageAllowed(ssh, MSGID_KEXDH_INIT, WS_MSG_RECV));
ssh->handshake->expectMsgId = MSGID_NEWKEYS;
CHECK(!IsMessageAllowed(ssh, MSGID_KEXDH_REPLY, WS_MSG_RECV));
CHECK(ssh->handshake->expectMsgId == MSGID_NEWKEYS);
CHECK(IsMessageAllowed(ssh, MSGID_IGNORE, WS_MSG_RECV));
CHECK(ssh->handshake->expectMsgId == MSGID_NEWKEYS);
}
ssh->ctx->side = WOLFSSH_ENDPOINT_SERVER;
ssh->isKeying = 0;
ssh->acceptState = ACCEPT_KEYED;
CHECK(IsMessageAllowed(ssh, MSGID_SERVICE_REQUEST, WS_MSG_RECV));
CHECK(!IsMessageAllowed(ssh, MSGID_USERAUTH_REQUEST, WS_MSG_RECV));
ssh->acceptState = ACCEPT_SERVER_USERAUTH_ACCEPT_SENT;
CHECK(IsMessageAllowed(ssh, MSGID_USERAUTH_REQUEST, WS_MSG_RECV));
CHECK(!IsMessageAllowed(ssh, MSGID_SERVICE_REQUEST, WS_MSG_RECV));
CHECK(!IsMessageAllowed(ssh, MSGID_USERAUTH_SUCCESS, WS_MSG_RECV));
CHECK(!IsMessageAllowed(ssh, MSGID_CHANNEL_OPEN, WS_MSG_RECV));
ssh->acceptState = ACCEPT_SERVER_USERAUTH_SENT;
CHECK(IsMessageAllowed(ssh, MSGID_CHANNEL_OPEN, WS_MSG_RECV));
CHECK(!IsMessageAllowed(ssh, MSGID_USERAUTH_REQUEST, WS_MSG_RECV));
CHECK(!IsMessageAllowed(ssh, MSGID_EXT_INFO, WS_MSG_RECV));
CHECK(SendExtInfo(ssh) == WS_NOT_COMPILED);
ssh->ctx->side = WOLFSSH_ENDPOINT_CLIENT;
ssh->connectState = CONNECT_KEYED;
CHECK(!IsMessageAllowed(ssh, MSGID_USERAUTH_FAILURE, WS_MSG_RECV));
CHECK(!IsMessageAllowed(ssh, MSGID_USERAUTH_SUCCESS, WS_MSG_RECV));
CHECK(!IsMessageAllowed(ssh, MSGID_CHANNEL_OPEN_CONF, WS_MSG_RECV));
ssh->connectState = CONNECT_CLIENT_USERAUTH_SENT;
CHECK(IsMessageAllowed(ssh, MSGID_USERAUTH_FAILURE, WS_MSG_RECV));
CHECK(IsMessageAllowed(ssh, MSGID_USERAUTH_SUCCESS, WS_MSG_RECV));
ssh->connectState = CONNECT_SERVER_USERAUTH_ACCEPT_DONE;
CHECK(!IsMessageAllowed(ssh, MSGID_USERAUTH_FAILURE, WS_MSG_RECV));
CHECK(IsMessageAllowed(ssh, MSGID_CHANNEL_OPEN_CONF, WS_MSG_RECV));
/* A wrong first_kex_packet_follows guess does not consume the real INIT. */
ssh->ctx->side = WOLFSSH_ENDPOINT_SERVER;
ssh->isKeying = WOLFSSH_PEER_IS_KEYING;
ssh->handshake->kexPacketFollows = 1;
ssh->handshake->kexIdGuess = ID_UNKNOWN;
ssh->handshake->kexId = ID_CURVE25519_SHA256;
ssh->handshake->expectMsgId = MSGID_KEXDH_INIT;
CHECK(IsMessageAllowed(ssh, MSGID_KEXDH_INIT, WS_MSG_RECV));
word32 idx = 0;
byte guessed[1] = {0};
CHECK(DoKexDhInit(ssh, guessed, sizeof(guessed), &idx) == WS_SUCCESS);
CHECK(idx == 1 && ssh->handshake->expectMsgId == MSGID_KEXDH_INIT);
CHECK(DoKexInit(ssh, guessed, sizeof(guessed), &idx) == WS_INVALID_STATE_E);
/* Real dispatch must fail before parsing malformed auth/channel payloads. */
for (side = WOLFSSH_ENDPOINT_SERVER; side <= WOLFSSH_ENDPOINT_CLIENT; ++side) {
ssh->ctx->side = side;
ssh->connectState = CONNECT_BEGIN;
for (msg = 0; msg < 256; ++msg) {
if (msg == MSGID_KEXINIT || (msg >= 1 && msg <= 4)) continue;
ssh->acceptState = ACCEPT_BEGIN;
ssh->isKeying = 0;
ssh->inputBuffer.idx = 0;
ssh->curSz = 8;
memset(ssh->inputBuffer.buffer, 0, 12);
ssh->inputBuffer.buffer[4] = 6;
ssh->inputBuffer.buffer[5] = msg;
CHECK(DoPacket(ssh, &consumed) == WS_MSGID_NOT_ALLOWED_E);
CHECK(consumed == 0);
}
}
ssh->ctx->side = WOLFSSH_ENDPOINT_SERVER;
}
static void newkeys(WOLFSSH_CTX* ctx)
{
int quota;
for (quota = 0; quota <= 16; ++quota) {
WOLFSSH* ssh = wolfSSH_new(ctx);
Sink sink = {.quota = quota, .chunk = 1};
CHECK(ssh != NULL);
CHECK(ssh->isKeying == 0);
wolfSSH_SetIOWriteCtx(ssh, &sink);
ssh->isKeying = WOLFSSH_SELF_IS_KEYING | WOLFSSH_PEER_IS_KEYING;
ssh->handshake->expectMsgId = MSGID_NEWKEYS;
ssh->handshake->encryptId = ID_AES128_GCM;
ssh->handshake->aeadMode = 1;
ssh->handshake->blockSz = 16;
ssh->handshake->keys.encKeySz = 16;
ssh->handshake->peerKeys.encKeySz = 16;
ssh->handshake->keys.ivSz = ssh->handshake->peerKeys.ivSz = 12;
CHECK(DoNewKeys(ssh, NULL, 0, NULL) == WS_INVALID_STATE_E);
int ret = SendNewKeys(ssh);
CHECK(ret == WS_SUCCESS || ret == WS_WANT_WRITE);
CHECK(ssh->isKeying == WOLFSSH_PEER_IS_KEYING);
CHECK(ssh->handshake->expectMsgId == MSGID_NEWKEYS);
word32 seq = ssh->seq;
sink.quota = -1;
CHECK(wolfSSH_SendPacket(ssh) == WS_SUCCESS);
CHECK(ssh->seq == seq);
CHECK(sink.bytes[5] == MSGID_NEWKEYS);
CHECK(sink.size == 16);
CHECK(ssh->outputBuffer.length == 0);
CHECK(IsMessageAllowed(ssh, MSGID_NEWKEYS, WS_MSG_RECV));
CHECK(DoNewKeys(ssh, NULL, 0, NULL) == WS_SUCCESS);
CHECK(ssh->isKeying == 0 && ssh->handshake == NULL);
CHECK(!IsMessageAllowed(ssh, MSGID_NEWKEYS, WS_MSG_RECV));
CHECK(DoNewKeys(ssh, NULL, 0, NULL) == WS_BAD_ARGUMENT);
wolfSSH_free(ssh);
}
WOLFSSH* ssh = wolfSSH_new(ctx);
Sink sink = {.fatal = 1, .chunk = 1};
wolfSSH_SetIOWriteCtx(ssh, &sink);
ssh->isKeying = WOLFSSH_SELF_IS_KEYING | WOLFSSH_PEER_IS_KEYING;
ssh->handshake->encryptId = ID_NONE;
CHECK(SendNewKeys(ssh) != WS_SUCCESS);
CHECK(ssh->isKeying & WOLFSSH_SELF_IS_KEYING);
wolfSSH_free(ssh);
ssh = wolfSSH_new(ctx);
CHECK(ssh != NULL);
wolfSSH_SetIOWriteCtx(ssh, &sink);
ssh->isKeying = WOLFSSH_SELF_IS_KEYING | WOLFSSH_PEER_IS_KEYING;
ssh->handshake->encryptId = ID_AES128_GCM;
ssh->handshake->keys.encKeySz = 15; /* invalid AES key length */
CHECK(SendNewKeys(ssh) != WS_SUCCESS);
CHECK(ssh->isKeying == (WOLFSSH_SELF_IS_KEYING | WOLFSSH_PEER_IS_KEYING));
ssh->isKeying = WOLFSSH_PEER_IS_KEYING;
ssh->handshake->peerKeys.encKeySz = 15;
CHECK(DoNewKeys(ssh, NULL, 0, NULL) == WS_CRYPTO_FAILED);
CHECK(ssh->handshake != NULL && ssh->isKeying == WOLFSSH_PEER_IS_KEYING);
wolfSSH_free(ssh);
}
static void shutdown_rekey(WOLFSSH_CTX* ctx)
{
const int errors[] = {WS_SUCCESS, WS_WANT_WRITE};
int keying, operation, pending, error;
for (keying = 1; keying <= 3; ++keying) {
for (operation = 0; operation < 3; ++operation) {
for (pending = 0; pending < 2; ++pending) {
for (error = 0; error < 2; ++error) {
WOLFSSH* ssh = wolfSSH_new(ctx);
Sink sink = {.quota = -1, .chunk = 1024};
CHECK(ssh != NULL);
wolfSSH_SetIOWriteCtx(ssh, &sink);
WOLFSSH_CHANNEL* channel = ChannelNew(ssh, ID_CHANTYPE_SESSION, 1024, 1024);
CHECK(channel != NULL);
channel->peerChannel = channel->channel;
CHECK(ChannelAppend(ssh, channel) == WS_SUCCESS);
if (pending) {
sink.quota = 0;
ssh->isKeying = WOLFSSH_SELF_IS_KEYING | WOLFSSH_PEER_IS_KEYING;
CHECK(SendNewKeys(ssh) == WS_WANT_WRITE);
sink.quota = -1;
}
ssh->isKeying = keying;
ssh->handshake->expectMsgId = MSGID_NEWKEYS;
word32 length = ssh->outputBuffer.length;
word32 index = ssh->outputBuffer.idx, seq = ssh->seq;
byte saved[64];
CHECK(length <= sizeof(saved));
memcpy(saved, ssh->outputBuffer.buffer, length);
int calls = sink.calls;
int repeat;
for (repeat = 0; repeat < 2; ++repeat) {
ssh->error = errors[error]; /* Including a stale WANT_WRITE. */
int ret = operation == 0 ? SendChannelEof(ssh, channel->peerChannel) :
operation == 1 ? wolfSSH_shutdown(ssh) :
SendChannelExit(ssh, channel->peerChannel, 0);
CHECK(ret == WS_MSGID_NOT_ALLOWED_E);
CHECK(ssh->error == WS_REKEYING);
CHECK(sink.size == 0 && sink.calls == calls);
CHECK(!channel->eofTxd && !channel->closeTxd);
CHECK(ssh->outputBuffer.length == length && ssh->outputBuffer.idx == index);
CHECK(memcmp(saved, ssh->outputBuffer.buffer, length) == 0);
CHECK(ssh->seq == seq && ssh->isKeying == keying);
CHECK(ssh->handshake->expectMsgId == MSGID_NEWKEYS);
}
/* Retry after keying: EOF is emitted once, not suppressed forever. */
ssh->isKeying = 0;
ssh->error = WS_SUCCESS;
CHECK(SendChannelEof(ssh, channel->peerChannel) == WS_SUCCESS);
CHECK(channel->eofTxd && !channel->closeTxd);
CHECK(sink.size == length + 16 && sink.bytes[length + 5] == MSGID_CHANNEL_EOF);
calls = sink.calls;
CHECK(SendChannelEof(ssh, channel->peerChannel) == WS_SUCCESS);
CHECK(sink.calls == calls);
wolfSSH_free(ssh);
}
}
}
}
puts("PASS: real EOF/shutdown/exit rekey fences, pending output/stale WANT_WRITE, zero bytes and no channel mutation");
}
#include "interop.c"
#include "paired.c"
int main(int argc, char** argv)
{
if (argc > 1) return serve(argc, argv);
CHECK(wolfSSH_Init() == WS_SUCCESS);
WOLFSSH_CTX* ctx = wolfSSH_CTX_new(WOLFSSH_ENDPOINT_SERVER, NULL);
CHECK(ctx != NULL);
policy(ctx);
wolfSSH_SetIOSend(ctx, send_test);
WOLFSSH* ssh = wolfSSH_new(ctx);
CHECK(ssh != NULL && ssh->handshake != NULL);
gates(ssh);
wolfSSH_free(ssh);
newkeys(ctx);
shutdown_rekey(ctx);
paired();
wolfSSH_CTX_free(ctx);
wolfSSH_Cleanup();
printf("PASS: %u real generated ordering/dispatch/NEWKEYS checks\n", checks);
return 0;
}
tests/wolfssh_order_contract/interop.c
+175
View File
@@ -0,0 +1,175 @@
/* Test-only stdio server; no sockets, persistent keys, or production auth. */
#include <fcntl.h>
#include <signal.h>
#include <execinfo.h>
static void trap_trace(int signum)
{
void* trace[32];
int count = backtrace(trace, 32);
backtrace_symbols_fd(trace, count, 2);
_exit(128 + signum);
}
static byte authorized[2048];
static size_t authorizedSz;
static int keyings, signedAuth, passwordAuth, writes, stalls, fragment, transportEof;
static int auth_test(byte type, WS_UserAuthData* data, void* context)
{
(void)context;
if (type == WOLFSSH_USERAUTH_PASSWORD &&
data->sf.password.passwordSz == 19 &&
memcmp(data->sf.password.password, "order-test-password", 19) == 0) {
++passwordAuth;
return WOLFSSH_USERAUTH_SUCCESS;
}
if (type == WOLFSSH_USERAUTH_PUBLICKEY &&
data->sf.publicKey.publicKeySz == authorizedSz &&
memcmp(data->sf.publicKey.publicKey, authorized, authorizedSz) == 0) {
if (data->sf.publicKey.hasSignature) ++signedAuth;
return WOLFSSH_USERAUTH_SUCCESS;
}
return WOLFSSH_USERAUTH_FAILURE;
}
static void keyed_test(void* context) { (void)context; ++keyings; }
static int stdio_recv(WOLFSSH* ssh, void* buffer, word32 size, void* context)
{
(void)ssh; (void)context;
if (fragment && size > 31) size = 31;
int ret = (int)read(0, buffer, size);
if (ret > 0) return ret;
if (ret < 0 && (errno == EAGAIN || errno == EINTR))
return WS_CBIO_ERR_WANT_READ;
if (ret == 0) transportEof = 1;
return WS_CBIO_ERR_CONN_CLOSE;
}
static int stdio_send(WOLFSSH* ssh, void* buffer, word32 size, void* context)
{
(void)ssh; (void)context;
if (fragment && ++writes % 3 == 0) {
++stalls;
return WS_CBIO_ERR_WANT_WRITE;
}
if (fragment && size > 37) size = 37;
int ret = (int)write(1, buffer, size);
if (ret > 0) return ret;
if (ret < 0 && (errno == EAGAIN || errno == EINTR))
return WS_CBIO_ERR_WANT_WRITE;
return WS_CBIO_ERR_GENERAL;
}
static int retry(int ret, WOLFSSH* ssh)
{
int error = ret == WS_FATAL_ERROR ? wolfSSH_get_error(ssh) : ret;
if (error == WS_WANT_READ || error == WS_WANT_WRITE || error == WS_REKEYING ||
error == WS_CHAN_RXD || error == WS_SUCCESS) return 1;
fprintf(stderr, "server failure ret=%d err=%d accept=%d keying=%d expected=%d\n",
ret, error, ssh->acceptState, ssh->isKeying,
ssh->handshake ? ssh->handshake->expectMsgId : -1);
return 0;
}
static int serve(int argc, char** argv)
{
byte key[4096], buffer[4096];
size_t keySz, sent = 0, length = 0, total = 0;
int triggered = 0, channelClosed = 0;
CHECK(argc == 6);
fragment = atoi(argv[4]);
int initiate = atoi(argv[5]);
alarm(30);
signal(SIGILL, trap_trace);
signal(SIGSEGV, trap_trace);
FILE* file = fopen(argv[2], "rb"); CHECK(file != NULL);
keySz = fread(key, 1, sizeof(key), file); fclose(file);
file = fopen(argv[3], "rb"); CHECK(file != NULL);
authorizedSz = fread(authorized, 1, sizeof(authorized), file); fclose(file);
CHECK(fcntl(0, F_SETFL, O_NONBLOCK) == 0);
CHECK(fcntl(1, F_SETFL, O_NONBLOCK) == 0);
CHECK(wolfSSH_Init() == 0);
WOLFSSH_CTX* ctx = wolfSSH_CTX_new(WOLFSSH_ENDPOINT_SERVER, NULL);
CHECK(ctx != NULL);
policy(ctx);
CHECK(wolfSSH_CTX_UsePrivateKey_buffer(ctx, key, keySz, WOLFSSH_FORMAT_ASN1) == 0);
wolfSSH_SetUserAuth(ctx, auth_test);
wolfSSH_SetKeyingCompletionCb(ctx, keyed_test);
wolfSSH_SetIORecv(ctx, stdio_recv);
wolfSSH_SetIOSend(ctx, stdio_send);
WOLFSSH* ssh = wolfSSH_new(ctx); CHECK(ssh != NULL);
int ret;
while ((ret = wolfSSH_accept(ssh)) != WS_SUCCESS) {
CHECK(retry(ret, ssh));
usleep(100);
}
CHECK(keyings == 1 && (signedAuth == 1 || passwordAuth == 1));
CHECK(ssh->sendExtInfo == 0 && ssh->extInfoSent == 0);
for (;;) {
if (ssh->outputBuffer.length) {
ret = wolfSSH_SendPacket(ssh);
CHECK(retry(ret, ssh));
if (ret != WS_SUCCESS) { usleep(100); continue; }
}
if (initiate && !triggered && total >= 65536 && !ssh->isKeying) {
triggered = 1;
ret = wolfSSH_TriggerKeyExchange(ssh);
CHECK(retry(ret, ssh));
}
if (ssh->isKeying) {
ret = wolfSSH_worker(ssh, NULL);
CHECK(retry(ret, ssh));
usleep(100);
continue;
}
if (sent < length) {
ret = wolfSSH_stream_send(ssh, buffer + sent, length - sent);
if (ret > 0) { sent += ret; total += ret; }
else { CHECK(retry(ret, ssh)); usleep(100); }
continue;
}
ret = wolfSSH_stream_read(ssh, buffer, sizeof(buffer));
if (ret == WS_EOF || (ssh->channelList && ssh->channelList->eofRxd)) break;
if (ret > 0) { length = ret; sent = 0; }
else { CHECK(retry(ret, ssh)); usleep(100); }
}
CHECK(total == 262144);
CHECK(keyings >= 2);
CHECK(!fragment || stalls > 0);
CHECK(ssh->sendExtInfo == 0 && ssh->extInfoSent == 0);
/* Queue shutdown once, then finish IO without re-enqueuing exit-status.
* WANT_READ is not channel-close completion. Keep the proxy read end alive
* until OpenSSH has also finished writing its transport disconnect. */
fragment = 0;
ret = wolfSSH_shutdown(ssh);
if (ret == WS_CHANNEL_CLOSED) channelClosed = 1;
else CHECK(retry(ret, ssh));
while (ssh->channelList || ssh->outputBuffer.length) {
ret = ssh->outputBuffer.length ? wolfSSH_SendPacket(ssh) :
wolfSSH_worker(ssh, NULL);
if (ret == WS_CHANNEL_CLOSED) {
CHECK(ssh->channelList == NULL);
channelClosed = 1;
}
else CHECK(retry(ret, ssh));
usleep(100);
}
CHECK(channelClosed && !transportEof);
while (!transportEof) {
ret = wolfSSH_worker(ssh, NULL);
if (transportEof) {
int error = ret == WS_FATAL_ERROR ? wolfSSH_get_error(ssh) : ret;
CHECK(error == WS_SOCKET_ERROR_E && ssh->isClosed && !ssh->connReset);
}
else CHECK(retry(ret, ssh));
usleep(100);
}
CHECK(!ssh->isKeying && !ssh->outputBuffer.length);
fprintf(stderr, "INTEROP PASS keyings=%d signed=%d password=%d stalls=%d bytes=%zu channel_closed=1 transport_eof=1\n",
keyings, signedAuth, passwordAuth, stalls, total);
wolfSSH_free(ssh);
wolfSSH_CTX_free(ctx);
wolfSSH_Cleanup();
return 0;
}
tests/wolfssh_order_contract/paired.c
+115
View File
@@ -0,0 +1,115 @@
/* Real generated client/server state machines over bounded nonblocking queues. */
typedef struct Pipe {
byte bytes[65536];
word32 read, length;
unsigned calls, stalls;
} Pipe;
static int pipe_send(WOLFSSH* ssh, void* data, word32 size, void* context)
{
Pipe* pipe = context;
(void)ssh;
if (++pipe->calls % 3 == 0) {
++pipe->stalls;
return WS_CBIO_ERR_WANT_WRITE;
}
if (size > 13) size = 13;
CHECK(size <= sizeof(pipe->bytes) - pipe->length);
memcpy(pipe->bytes + pipe->length, data, size);
pipe->length += size;
return size;
}
static int pipe_recv(WOLFSSH* ssh, void* data, word32 size, void* context)
{
Pipe* pipe = context;
(void)ssh;
word32 available = pipe->length - pipe->read;
if (!available) return WS_CBIO_ERR_WANT_READ;
if (size > available) size = available;
if (size > 11) size = 11;
memcpy(data, pipe->bytes + pipe->read, size);
pipe->read += size;
if (pipe->read == pipe->length) pipe->read = pipe->length = 0;
return size;
}
static int client_auth(byte type, WS_UserAuthData* data, void* context)
{
(void)context;
if (type != WOLFSSH_USERAUTH_PASSWORD) return WOLFSSH_USERAUTH_FAILURE;
data->sf.password.password = (const byte*)"order-test-password";
data->sf.password.passwordSz = 19;
return WOLFSSH_USERAUTH_SUCCESS;
}
static void paired(void)
{
const char* algorithms[] = {"curve25519-sha256", "ecdh-sha2-nistp256"};
int algorithm;
for (algorithm = 0; algorithm < 2; ++algorithm) {
WOLFSSH_CTX* serverCtx = wolfSSH_CTX_new(WOLFSSH_ENDPOINT_SERVER, NULL);
WOLFSSH_CTX* clientCtx = wolfSSH_CTX_new(WOLFSSH_ENDPOINT_CLIENT, NULL);
CHECK(serverCtx && clientCtx);
policy(serverCtx); policy(clientCtx);
CHECK(wolfSSH_CTX_SetAlgoListKex(clientCtx, algorithms[algorithm]) == 0);
CHECK(wolfSSH_CTX_SetAlgoListCipher(clientCtx, "aes256-gcm@openssh.com") == 0);
WC_RNG rng;
ecc_key key;
byte der[1024];
CHECK(wc_InitRng(&rng) == 0);
CHECK(wc_ecc_init(&key) == 0);
CHECK(wc_ecc_make_key(&rng, 32, &key) == 0);
int derSz = wc_EccKeyToDer(&key, der, sizeof(der));
CHECK(derSz > 0);
CHECK(wolfSSH_CTX_UsePrivateKey_buffer(serverCtx, der, derSz, WOLFSSH_FORMAT_ASN1) == 0);
wc_ecc_free(&key); wc_FreeRng(&rng);
wolfSSH_SetUserAuth(serverCtx, auth_test);
wolfSSH_SetUserAuth(clientCtx, client_auth);
wolfSSH_SetIOSend(serverCtx, pipe_send); wolfSSH_SetIORecv(serverCtx, pipe_recv);
wolfSSH_SetIOSend(clientCtx, pipe_send); wolfSSH_SetIORecv(clientCtx, pipe_recv);
WOLFSSH* server = wolfSSH_new(serverCtx);
WOLFSSH* client = wolfSSH_new(clientCtx);
Pipe toServer = {0}, toClient = {0};
CHECK(server && client);
CHECK(wolfSSH_SetUsername(client, "order-test") == 0);
wolfSSH_SetIOWriteCtx(client, &toServer); wolfSSH_SetIOReadCtx(server, &toServer);
wolfSSH_SetIOWriteCtx(server, &toClient); wolfSSH_SetIOReadCtx(client, &toClient);
int clientDone = 0, serverDone = 0, i, ret;
for (i = 0; i < 100000 && !(clientDone && serverDone); ++i) {
if (!clientDone) {
ret = wolfSSH_connect(client);
if (ret == 0) clientDone = 1;
else CHECK(retry(ret, client));
}
if (!serverDone) {
ret = wolfSSH_accept(server);
if (ret == 0) serverDone = 1;
else CHECK(retry(ret, server));
}
}
CHECK(clientDone && serverDone);
CHECK(client->handshake == NULL && server->handshake == NULL);
CHECK(client->sendExtInfo == 0 && server->sendExtInfo == 0);
CHECK(client->peerSigId == NULL && client->peerSigIdSz == 0);
/* Both initiation directions and simultaneous initiation. */
int round;
for (round = 0; round < 3; ++round) {
if (round != 1) CHECK(retry(wolfSSH_TriggerKeyExchange(server), server));
if (round != 0) CHECK(retry(wolfSSH_TriggerKeyExchange(client), client));
for (i = 0; i < 100000; ++i) {
CHECK(retry(wolfSSH_worker(server, NULL), server));
CHECK(retry(wolfSSH_worker(client, NULL), client));
if (!server->isKeying && !client->isKeying &&
!toServer.length && !toClient.length) break;
}
CHECK(i < 100000);
CHECK(client->handshake == NULL && server->handshake == NULL);
CHECK(client->sendExtInfo == 0 && server->sendExtInfo == 0);
}
CHECK(toServer.stalls > 0 && toClient.stalls > 0);
wolfSSH_free(client); wolfSSH_free(server);
wolfSSH_CTX_free(clientCtx); wolfSSH_CTX_free(serverCtx);
}
puts("PASS: real generated client/server, both KEX, AES256, fragmented initial KEX and three rekey directions");
}
tests/wolfssh_order_contract/pio_adapter.py
+105
View File
@@ -0,0 +1,105 @@
#!/usr/bin/env python3
"""Exercise installed PlatformIO ESP-IDF flag sorting and SCons deduplication.
Uses the real configured CMake file API and target compiler. Never runs pio or
writes build artifacts; compiler outputs and the split-option mutation are temp.
"""
import ast
import json
import os
from pathlib import Path
import shlex
import subprocess
import sys
import tempfile
import click
ROOT = Path(__file__).resolve().parents[2]
def main():
platformio = Path.home() / '.platformio'
adapter = platformio / 'platforms/espressif32/builder/frameworks/espidf.py'
tree = ast.parse(adapter.read_text())
functions = [node for node in tree.body if isinstance(node, ast.FunctionDef)
and node.name == 'get_app_flags']
assert len(functions) == 1
scope = {'click': click}
exec(compile(ast.Module(body=functions, type_ignores=[]), str(adapter), 'exec'), scope)
scons = list((platformio / 'packages/tool-scons').glob('scons-local-*'))
assert len(scons) == 1
sys.path.insert(0, str(scons[0]))
from SCons.Script import Environment
build = ROOT / '.pio/build/esp32-s3-devkitc-1-n16r8'
replies = build / '.cmake/api/v1/reply'
targets = [json.loads(path.read_text()) for path in replies.glob('target-*.json')]
matches = [target for target in targets if target['name'] == '__idf_src']
assert len(matches) == 1, 'requires configured PlatformIO app target'
app = matches[0]
groups = [group for group in app['compileGroups'] if group['language'] == 'C']
assert len(groups) == 1
group = groups[0]
defaults = {'compileGroups': [{'language': lang, 'compileCommandFragments': []}
for lang in ('C', 'CXX', 'ASM')]}
commands = json.loads((build / 'compile_commands.json').read_text())
main_command = next(entry for entry in commands if entry['file'].endswith('/src/main.c'))
compiler = (main_command.get('arguments') or shlex.split(main_command['command']))[0]
overlay = str(build / 'security_overrides/wolfssh_include/wolfssh/internal.h')
crypto = str(ROOT / 'cmake/wolf_crypto_policy.h')
defines = ['-D' + item['define'] for item in group.get('defines', [])]
includes = ['-I' + item['path'] for item in group.get('includes', [])]
flags = scope['get_app_flags'](app, defaults)['CFLAGS']
for header in (overlay, crypto):
assert '-include' + header in flags, header
assert header not in flags, 'orphan header operand'
assert '-include' not in flags
env = Environment(tools=[])
env.AppendUnique(CFLAGS=flags)
env.AppendUnique(CFLAGS=flags)
assert list(env['CFLAGS']) == flags
# Component path uses ParseFlags/AppendUnique rather than get_app_flags.
component_env = Environment(tools=[])
for header in (crypto, overlay, crypto, overlay):
component_env.AppendUnique(**component_env.ParseFlags('-include' + header))
assert list(component_env['CCFLAGS']) == ['-include' + crypto, '-include' + overlay]
with tempfile.TemporaryDirectory(prefix='pio-forced-include-') as temp:
temp = Path(temp)
source = temp / 'consumer.c'
source.write_text('''#if SAK_WOLFSSH_ORDER_ABI != 20260916
#error missing_shared_ssh_ABI
#endif
#if !defined(WOLFSSL_VALIDATE_ECC_IMPORT) || !defined(WOLFSSL_ECDHX_SHARED_NOT_ZERO)
#error missing_crypto_guards
#endif
int consumer(void) { return 0; }
''')
command = [compiler, *env['CFLAGS'], *defines, *includes, '-c', str(source), '-o', str(temp / 'consumer.o')]
result = subprocess.run(command, cwd=ROOT, capture_output=True, text=True, timeout=30,
env={**os.environ, 'CCACHE_DISABLE': '1'})
assert result.returncode == 0, result.stderr
assert (temp / 'consumer.o').is_file()
# Prove this catches the historical split-option failure using the same
# installed adapter, not a hand-written approximation of its sorting.
bad = json.loads(json.dumps(app))
count = 0
for cg in bad['compileGroups']:
for fragment in cg['compileCommandFragments']:
text = fragment['fragment']
if '-include' + overlay in text:
fragment['fragment'] = text.replace('-include' + overlay, '-include ' + overlay)
count += 1
assert count > 0
broken_flags = scope['get_app_flags'](bad, defaults)['CFLAGS']
result = subprocess.run([compiler, *broken_flags, *defines, *includes, '-c', str(source),
'-o', str(temp / 'broken.o')], cwd=ROOT, capture_output=True,
text=True, timeout=30, env={**os.environ, 'CCACHE_DISABLE': '1'})
assert result.returncode != 0 and 'multiple files' in result.stderr, result.stderr
print('PASS: installed PlatformIO get_app_flags + SCons AppendUnique/ParseFlags; '
'real Xtensa -c/-o consumer; split-include mutation reproduces multiple-input failure')
if __name__ == '__main__':
main()
tests/wolfssh_order_contract/run.py
+314
View File
@@ -0,0 +1,314 @@
#!/usr/bin/env python3
"""Offline real generated wolfSSH/real wolfCrypt ordering and interoperability tests.
No PlatformIO, device, managed writes, or IP network. OpenSSH uses a local Unix socket.
"""
import argparse
import array
import socket
import hashlib
import base64
import shlex
import resource
import re
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
sys.dont_write_bytecode = True
ROOT = Path(__file__).resolve().parents[2]
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT / 'tools'))
import security_overrides as sdk
SSH = ROOT / 'managed_components/wolfssl__wolfssh'
SSL = ROOT / 'managed_components/wolfssl__wolfssl'
ENV = {**os.environ, 'CCACHE_DISABLE': '1'}
def run(args, **kw):
return subprocess.run([str(x) for x in args], check=True, timeout=120,
env=ENV, **kw)
def provenance():
directory = ROOT / 'tools/wolfssh_order'
records = json.loads((directory / 'provenance.json').read_text())
assert set(records) == {'793', '819', '840', '855', '921'}
for number, record in records.items():
raw = (directory / (number + '.patch')).read_bytes()
assert hashlib.sha256(raw).hexdigest() == record['sha256']
assert re.findall(rb'^From ([0-9a-f]{40}) Mon Sep', raw, re.M) == [
commit.encode() for commit in record['commits']]
assert set(sdk.WOLFSSH_ORDER_PLAN) == {'src/internal.c', 'src/ssh.c', 'wolfssh/internal.h'}
for path, record in sdk.WOLFSSH_ORDER_PLAN.items():
entry = next(e for e in sdk.ENTRIES if e.source == 'managed_components/wolfssl__wolfssh/' + path)
assert record['sha256'] == entry.sha256
assert hashlib.sha256((SSH / 'src/log.c').read_bytes()).hexdigest() == \
'66e5f053af05a103aca997fbcae11a52acd8f79c9c97645f0d0cd390e192135a'
assert hashlib.sha256((SSH / 'wolfssh/log.h').read_bytes()).hexdigest() == \
'4a3e71f8148b0e6ecd533b949e2f9574449561bda21bc200163a1e53bfdb0bd2'
print('PASS: archived upstream patch hashes/commits; exact source/header pins; unchanged logging ABI')
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--interop', action='store_true')
parser.add_argument('--interop-repeat', type=int, default=1)
parser.add_argument('--target-contracts', action='store_true')
parser.add_argument('--pio-adapter', action='store_true')
parser.add_argument('--proxy-socket', type=Path, help=argparse.SUPPRESS)
options = parser.parse_args()
if options.proxy_socket:
with socket.socket(socket.AF_UNIX) as connection:
connection.connect(str(options.proxy_socket))
with socket.socket(fileno=1) as output:
output.sendmsg([b'\0'], [(socket.SOL_SOCKET, socket.SCM_RIGHTS,
array.array('i', [connection.fileno()]))])
return
if not 1 <= options.interop_repeat <= 20:
parser.error('--interop-repeat must be between 1 and 20')
provenance()
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
with tempfile.TemporaryDirectory(prefix='wolfssh-order-') as tmp:
work = Path(tmp)
for entry in sdk.ENTRIES:
if entry.component != 'wolfssl__wolfssh':
continue
_, data = sdk.render_entry(entry, {'project': ROOT})
dest = (work / 'wolfssh/internal.h' if entry.header else
work / Path(entry.source).name)
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_bytes(data)
flags = ['cc', '-std=gnu11', '-O1', '-g', '-DWOLFSSL_USER_SETTINGS',
'-DHAVE_WC_ECC_SET_RNG', '-ffunction-sections', '-fdata-sections',
'-fsanitize=undefined', '-fsanitize-undefined-trap-on-error', '-no-pie',
'-I' + str(HERE), '-I' + str(SSH), '-I' + str(work), '-I' + str(SSL),
'-include', str(work / 'wolfssh/internal.h')]
stale = subprocess.run([*flags[:-2], '-include', str(SSH / 'wolfssh/internal.h'),
*flags[-2:], '-x', 'c', '-fsyntax-only', '-'],
input='', text=True, capture_output=True, timeout=30, env=ENV)
assert stale.returncode != 0 and 'stale wolfSSH internal.h' in stale.stderr
print('PASS: vendor-first forced-header ABI conflict rejected')
crypto = ('aes asn coding curve25519 ecc ed25519 fe_low_mem ge_low_mem '
'hash hmac kdf logging memory random sha256 sha512 signature '
'tfm wc_port wolfmath').split()
sources = [work / 'ssh.c', *[SSH / 'src' / (s + '.c') for s in ('io', 'log', 'port')],
*[SSL / 'wolfcrypt/src' / (s + '.c') for s in crypto]]
objects = []
for i, source in enumerate(sources):
obj = work / f'{i}.o'
run([*flags, '-c', source, '-o', obj], capture_output=True)
objects.append(obj)
binary = work / 'contract'
run([*flags, HERE / 'contract.c', *objects, '-Wl,--gc-sections', '-o', binary])
run([binary])
mutations(work, flags, objects)
if options.interop:
for iteration in range(options.interop_repeat):
print(f'INTEROP matrix {iteration + 1}/{options.interop_repeat}', flush=True)
interop(work, binary)
if options.target_contracts:
target_contracts(work)
if options.pio_adapter:
run([sys.executable, HERE / 'pio_adapter.py'])
print('PASS: generated pinned source/header ordering profile (host, not firmware)')
def target_contracts(work):
"""Explicit candidate replay, not a firmware reconfigure/build claim."""
original_db = ROOT / '.pio/build/esp32-s3-devkitc-1-n16r8/compile_commands.json'
entries = json.loads(original_db.read_text())
replacements = {}
for entry in sdk.ENTRIES:
if entry.component != 'wolfssl__wolfssh':
continue
_, data = sdk.render_entry(entry, {'project': ROOT})
dest = (work / 'security_overrides/wolfssh_include/wolfssh/internal.h' if entry.header
else work / 'security_overrides' / entry.name / Path(entry.source).name)
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_bytes(data)
if entry.header:
overlay = dest
else:
replacements[entry.name] = dest
for entry in entries:
source = Path(entry['file'])
if not source.is_absolute():
source = Path(entry['directory']) / source
args = entry.get('arguments') or shlex.split(entry['command'])
dest = None
if source.name == 'internal.c' and 'wolfssh_internal' in source.parts:
dest = replacements['wolfssh_internal']
if source.name == 'ssh.c' and ('wolfssl__wolfssh' in source.parts or 'wolfssh_ssh' in source.parts):
dest = replacements['wolfssh_ssh']
if dest:
args = [str(dest) if arg == entry['file'] or arg == str(source) else arg for arg in args]
entry['file'] = str(dest)
args = [(('-include' if arg.startswith('-include') else '') + str(overlay))
if arg.endswith('/security_overrides/wolfssh_include/wolfssh/internal.h')
else arg for arg in args]
# Same ABI overlay for all candidate consumers. Original compile flags,
# crypto policy and feature settings are preserved, not synthesized.
if any(arg.startswith('-I') and 'wolfssl__wolfssh' in arg for arg in args):
args += ['-include' + str(overlay)]
entry['arguments'] = args
entry.pop('command', None)
database = work / 'candidate-compile_commands.json'
database.write_text(json.dumps(entries))
print('CANDIDATE ONLY: replaying existing target flags with freshly generated SSH sources/header; '
'no production build-tree changes', flush=True)
syntax_sources = []
for entry in entries:
if (entry['file'] not in map(str, replacements.values()) and
not entry['file'].endswith(('/src/ssh_transport.c', '/src/ssh_security.c'))):
continue
command = []
skip = False
for arg in entry['arguments']:
if skip:
skip = False
elif arg in ('-o', '-MF', '-MT', '-MQ'):
skip = True
elif arg not in ('-c', '-MD', '-MMD', '-MP'):
command.append(arg)
run([*command, '-fsyntax-only'], cwd=entry['directory'], capture_output=True)
syntax_sources.append(entry['file'])
assert len(syntax_sources) == len(set(syntax_sources)) == 4, syntax_sources
print('PASS: candidate Xtensa syntax for generated internal.c/ssh.c and SSH application consumers')
for suite in ('ssh_protocol_policy', 'wolf_crypto_policy', 'wolfssh_auth_contract'):
run([sys.executable, ROOT / 'tests' / suite / 'run.py', '--compile-commands', database])
print('PASS: candidate target source-contract replay; NOT firmware/build registration evidence')
def mutations(work, flags, objects):
source = work / 'internal.c'
original = source.read_text()
cases = (
('missing EOF rekey guard',
' if (ret == WS_SUCCESS) {\n'
' if (!IsMessageAllowed(ssh, MSGID_CHANNEL_EOF, WS_MSG_SEND)) {\n'
' ret = WS_MSGID_NOT_ALLOWED_E;\n }\n }\n\n', ''),
('preauth injection', ' if (state != WS_MSG_RECV)\n',
' if (msg == MSGID_USERAUTH_FAILURE) return 1;\n if (state != WS_MSG_RECV)\n'),
('wrong expected KEX', 'ssh->handshake->expectMsgId == msg)',
'ssh->handshake->expectMsgId != msg)'),
('unnegotiated EXT_INFO', 'if (msg == MSGID_EXT_INFO)\n goto reject;',
'if (msg == MSGID_EXT_INFO)\n return 1;'),
('NEWKEYS clears peer too', 'ssh->isKeying &= ~WOLFSSH_SELF_IS_KEYING;',
'ssh->isKeying = 0;'),
('peer NEWKEYS before local', '(ssh->isKeying & WOLFSSH_SELF_IS_KEYING) ||\n ', ''),
('PR921 missing server expectation',
' if (ret == WS_SUCCESS) {\n ssh->handshake->expectMsgId = MSGID_NEWKEYS;\n'
' WLOG_EXPECT_MSGID(ssh->handshake->expectMsgId);\n ret = SendNewKeys(ssh);\n }\n\n'
' if (ret != WS_WANT_WRITE && ret != WS_SUCCESS)',
' if (ret == WS_SUCCESS) {\n ret = SendNewKeys(ssh);\n }\n\n'
' if (ret != WS_WANT_WRITE && ret != WS_SUCCESS)'),
)
try:
for label, old, new in cases:
assert original.count(old) == 1, label
source.write_text(original.replace(old, new))
binary = work / 'mutation'
run([*flags, HERE / 'contract.c', *objects, '-Wl,--gc-sections', '-o', binary], capture_output=True)
result = subprocess.run([str(binary)], capture_output=True, timeout=30, env=ENV)
assert result.returncode != 0, 'undetected ordering mutation: ' + label
finally:
source.write_text(original)
print(f'PASS: {len(cases)} ordering/NEWKEYS/PR921 guard-removal mutations rejected')
def interop(work, binary):
run(['openssl', 'ecparam', '-name', 'prime256v1', '-genkey', '-noout',
'-outform', 'DER', '-out', work / 'host.der'], capture_output=True)
for kind in ('ed25519', 'ecdsa'):
(work / kind).unlink(missing_ok=True)
(work / (kind + '.pub')).unlink(missing_ok=True)
run(['ssh-keygen', '-q', '-t', kind, '-N', '', '-C', 'order-test',
'-f', work / kind], capture_output=True)
askpass = work / 'askpass'
askpass.write_text('#!/bin/sh\nprintf "%s\\n" "order-test-password"\n')
askpass.chmod(0o700)
payload = bytes(range(256)) * 1024
for kex in ('curve25519-sha256', 'ecdh-sha2-nistp256'):
for auth in ('ed25519', 'ecdsa', 'password'):
key = work / ('ecdsa' if auth == 'password' else auth)
(work / 'authorized').write_bytes(base64.b64decode(
key.with_suffix('.pub').read_text().split()[1]))
for fragment, initiate in ((0, 0), (1, 1)):
endpoint = work / 'interop.sock'
proxy = ' '.join(shlex.quote(str(x)) for x in
(sys.executable, HERE / 'run.py', '--proxy-socket', endpoint))
server_args = [str(x) for x in
(binary, '--stdio', work / 'host.der', work / 'authorized', fragment, initiate)]
args = ['ssh', '-F', '/dev/null', '-T', '-vv', '-o', 'StrictHostKeyChecking=no',
'-o', 'UserKnownHostsFile=/dev/null', '-o', 'GlobalKnownHostsFile=/dev/null',
'-o', 'IdentityAgent=none', '-o', 'IdentitiesOnly=yes',
'-o', 'HostKeyAlgorithms=ecdsa-sha2-nistp256',
'-o', 'KexAlgorithms=' + kex, '-o', 'Ciphers=aes128-gcm@openssh.com',
'-o', 'RekeyLimit=' + ('1G' if initiate else '32K'),
'-o', 'ProxyCommand=' + proxy, '-o', 'ProxyUseFdpass=yes',
'-o', 'NumberOfPasswordPrompts=1',
'-o', 'PreferredAuthentications=' + ('password' if auth == 'password' else 'publickey'),
'-i', str(key), 'order-test@stdio.invalid']
env = {**ENV, 'SSH_ASKPASS': str(askpass), 'SSH_ASKPASS_REQUIRE': 'force', 'DISPLAY': ':order-test'}
# Own/reap the server independently: OpenSSH kills its proxy at
# exit. The proxy only passes a local Unix socket, not the server.
with socket.socket(socket.AF_UNIX) as listener, tempfile.TemporaryFile() as server_log:
listener.bind(str(endpoint))
listener.listen(1)
listener.settimeout(10)
client = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, env=env)
server = None
try:
connection, _ = listener.accept()
with connection:
server = subprocess.Popen(server_args, stdin=connection, stdout=connection,
stderr=server_log, env=env)
stdout, stderr = client.communicate(payload, timeout=45)
server_rc = server.wait(timeout=5)
finally:
for process in (client, server):
if process is not None and process.poll() is None:
process.kill()
process.wait(timeout=5)
endpoint.unlink(missing_ok=True)
server_log.seek(0)
server_text = server_log.read().decode(errors='replace')
result = subprocess.CompletedProcess(args, client.returncode, stdout, stderr)
client_log = stderr.decode(errors='replace')
log = client_log + '\nSERVER:\n' + server_text
evidence = re.fullmatch(
r'INTEROP PASS keyings=(\d+) signed=(\d+) password=(\d+) stalls=(\d+) '
r'bytes=262144 channel_closed=1 transport_eof=1\n', server_text)
valid = (evidence is not None and int(evidence[1]) >= 2 and
(int(evidence[2]), int(evidence[3])) ==
((0, 1) if auth == 'password' else (1, 0)) and
(not fragment or int(evidence[4]) > 0) and
'rtype exit-status reply 0' in client_log and
'channel 0: rcvd close' in client_log and
'Exit status 0' in client_log)
if result.returncode or server_rc or result.stdout != payload or not valid:
addresses = re.findall(r'\[(0x[0-9a-f]+)\]', log)
if addresses:
run(['addr2line', '-f', '-e', binary, *addresses])
raise AssertionError(f'{kex}/{auth}/{fragment}/{initiate}: rc={result.returncode} server_rc={server_rc} '
f'output={len(result.stdout)}\n{log}')
assert 'SSH2_MSG_EXT_INFO received' not in log
assert 'server-sig-algs=<' not in log
print(f'PASS: OpenSSH {kex}/{auth} fragment={fragment} server-rekey={initiate}; '
+ next(line for line in log.splitlines() if 'INTEROP PASS' in line))
if __name__ == '__main__':
try:
main()
except subprocess.CalledProcessError as error:
if error.stdout:
print(error.stdout.decode() if isinstance(error.stdout, bytes) else error.stdout)
if error.stderr:
print(error.stderr.decode() if isinstance(error.stderr, bytes) else error.stderr)
raise
tests/wolfssh_order_contract/user_settings.h
+43
View File
@@ -0,0 +1,43 @@
/* Host test profile: real generated wolfSSH + real wolfCrypt, not ESP-IDF. */
#ifndef ORDER_TEST_SETTINGS_H
#define ORDER_TEST_SETTINGS_H
#include <strings.h>
#define WOLFCRYPT_ONLY
#define WOLFSSL_WOLFSSH
/* Use the portable bytewise encoding path under alignment sanitization. */
#define WOLFSSL_USE_ALIGN
#define SINGLE_THREADED
#define USE_FAST_MATH
#define TFM_NO_ASM
#define TFM_TIMING_RESISTANT
#define WOLFSSL_SMALL_STACK
#define NO_RSA
#define NO_DH
#define NO_DSA
#define NO_MD5
#define NO_SHA
#define NO_SHA224
#define NO_DES3
#define NO_RC4
#define NO_HC128
#define NO_RABBIT
#define NO_PSK
#define NO_PWDBASED
#define NO_PKCS12
#define NO_CERTS
#define WOLFSSL_ASN_TEMPLATE
#define HAVE_ECC
#define ECC_TIMING_RESISTANT
#define HAVE_ECC_CHECK_KEY
#define WOLFSSL_VALIDATE_ECC_IMPORT
#define HAVE_CURVE25519
#define CURVE25519_SMALL
#define WOLFSSL_ECDHX_SHARED_NOT_ZERO
#define HAVE_ED25519
#define ED25519_SMALL
#define WOLFSSL_SHA512
#define WOLFSSL_ED25519_STREAMING_VERIFY
#define HAVE_AESGCM
#define WOLFSSH_NO_AES_CBC
#define WOLFSSH_NO_AES_CTR
#endif
tests/wolfssh_parser_contract/run.py
+4 -3
View File
@@ -20,9 +20,10 @@ original, generated = render_entry(entry, {'project': ROOT})
names = ('GetUint32', 'GetSize', 'GetString', 'GetSkip', 'GetStringRef',
'DoIgnore', 'DoServiceRequest', 'DoChannelWindowAdjust', 'DoUserAuthRequestEcc',
'DoUserAuthRequestEd25519')
# Verify this slice cannot accidentally change ordering or existing password logic.
# Parser edits must not change the independently applied ordering/password logic.
from security_overrides import apply_edits, MODIFICATION_NOTICE, WOLFSSH_PARSER_EDITS
baseline = MODIFICATION_NOTICE + apply_edits(original.read_text(), entry.edits[len(WOLFSSH_PARSER_EDITS):])
baseline = MODIFICATION_NOTICE + apply_edits(original.read_text(), tuple(
edit for edit in entry.edits if edit not in WOLFSSH_PARSER_EDITS))
for name in ('DoUserAuthRequestPassword', 'DoPacket', 'DoChannelFailure',
'ParseRSAPubKey', 'ParseECCPubKey', 'DoUserAuthRequestPublicKey'):
assert extract(generated.decode(), name) == extract(baseline, name), name
@@ -75,5 +76,5 @@ with tempfile.TemporaryDirectory(prefix='wolfssh-parser-') as directory:
result = subprocess.run([str(binary)], capture_output=True, timeout=30)
assert result.returncode != 0, f'Undetected mutation: {label}'
print(f'PASS: {len(mutations)} parser guard-removal mutations rejected')
print('PASS: exact original hash; generated parser; unchanged ordering/password/deferred functions')
print('PASS: exact original hash; generated parser; parser-isolated ordering/password/deferred functions')
print('NOTE: production build-tree registration/firmware not regenerated or validated')