anti/DECNET
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(security): close INFO ASVS findings — secret echo, TLS floor, mandatory tarball SHA, CORS/Content-Type guards, BUG-17

Browse Source 
- V7.1.3: env known-insecure-default error no longer echoes the rejected secret value.
- V9.1.4: syslog-over-TLS forwarder + listener pin minimum_version=TLSv1_2.
- V12.1.2: updater tarball SHA-256 verification is now mandatory and fail-closed —
  /update and /update-self reject a missing digest (400), the executor rejects
  missing/mismatched digests before extract/apply. Every push path supplies it.
- V13.1.4: reject a wildcard '*' in DECNET_CORS_ORIGINS at startup.
- V13.1.5: enforce application/json on JSON write endpoints (415 otherwise),
  exempting multipart upload routes.
- BUG-17: SSE error log records the user uuid, not the resume cursor.

Also completes V2.1.7 consistently: the attacker-injectable PYTEST* env bypass is
replaced with explicit DECNET_TESTING=1 in the three remaining sites
(env.validate_public_binding, config logging, mysql url builder).

Tests added for every fix; unanimous adversarial review (no update-outage risk —
all push paths verified to send the digest).
This commit is contained in:
anti
2026-06-10 13:50:06 -04:00
parent 245975a6dd
commit 337520c7ad
17 changed files with 520 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

204
tests/api/test_security_middleware.py Normal file
View File

@@ -0,0 +1,204 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Security-middleware tests covering:
- V13.1.4: CORS wildcard guard raises ValueError at app startup
- V13.1.5: Content-Type enforcement middleware (415 on wrong CT; pass for
application/json; multipart exempt paths; GET/DELETE unaffected)
- BUG-17: SSE stream error log uses user["uuid"], not last_event_id
- Regression: multipart upload endpoints still work (canary blob, file-drop)
"""
from __future__ import annotations
import pytest
import httpx
# ---------------------------------------------------------------------------
# V13.1.4 — CORS wildcard guard (unit tests; lifespan path in tests/web/)
# ---------------------------------------------------------------------------
class TestCORSWildcardGuard:
def test_wildcard_raises(self):
"""_check_cors_origins raises ValueError when '*' is present."""
from decnet.web.api import _check_cors_origins
with pytest.raises(ValueError, match="wildcard"):
_check_cors_origins(["*"])
def test_wildcard_among_explicit_origins_raises(self):
"""Wildcard in a mixed list is still rejected."""
from decnet.web.api import _check_cors_origins
with pytest.raises(ValueError, match="wildcard"):
_check_cors_origins(["https://example.com", "*"])
def test_explicit_origins_ok(self):
"""Explicit origin URLs pass without raising."""
from decnet.web.api import _check_cors_origins
_check_cors_origins(["https://example.com", "https://app.internal"])
def test_empty_origins_ok(self):
"""Empty list is valid (no CORS)."""
from decnet.web.api import _check_cors_origins
_check_cors_origins([])
# ---------------------------------------------------------------------------
# V13.1.5 — Content-Type enforcement middleware
# ---------------------------------------------------------------------------
class TestContentTypeMiddleware:
@pytest.mark.asyncio
async def test_post_wrong_content_type_returns_415(
self, client: httpx.AsyncClient
):
"""POST with text/plain body to a JSON endpoint returns 415.
/api/v1/auth/login is the most stable JSON POST target — no auth
required, always present, middleware fires before the handler.
"""
resp = await client.post(
"/api/v1/auth/login",
content=b"not json",
headers={"Content-Type": "text/plain"},
)
assert resp.status_code == 415
@pytest.mark.asyncio
async def test_post_application_json_passes_middleware(
self, client: httpx.AsyncClient
):
"""POST with application/json does NOT get a 415 from middleware."""
resp = await client.post(
"/api/v1/auth/login",
json={"username": "nobody", "password": "wrong"},
)
# Middleware passes; handler may 401/422 but must not 415.
assert resp.status_code != 415
@pytest.mark.asyncio
async def test_post_json_with_charset_passes(
self, client: httpx.AsyncClient
):
"""application/json; charset=utf-8 is a valid Content-Type."""
resp = await client.post(
"/api/v1/auth/login",
content=b'{"username":"x","password":"y"}',
headers={"Content-Type": "application/json; charset=utf-8"},
)
assert resp.status_code != 415
@pytest.mark.asyncio
async def test_get_not_enforced(self, client: httpx.AsyncClient, auth_token: str):
"""GET requests are never rejected by the CT middleware."""
resp = await client.get(
"/api/v1/logs",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert resp.status_code != 415
@pytest.mark.asyncio
async def test_delete_not_enforced(
self, client: httpx.AsyncClient, auth_token: str
):
"""DELETE requests are never rejected by the CT middleware."""
resp = await client.delete(
"/api/v1/deckies/nonexistent",
headers={"Authorization": f"Bearer {auth_token}"},
)
# Could be 404/401/403 but never 415.
assert resp.status_code != 415
@pytest.mark.asyncio
async def test_multipart_canary_blob_exempt(
self, client: httpx.AsyncClient, auth_token: str
):
"""Canary blob upload (multipart/form-data) is NOT rejected with 415."""
resp = await client.post(
"/api/v1/canary/blobs",
files={"file": ("test.txt", b"hello world", "text/plain")},
headers={"Authorization": f"Bearer {auth_token}"},
)
# 201 on success, 4xx on business-logic errors — never 415.
assert resp.status_code != 415
@pytest.mark.asyncio
async def test_multipart_file_drop_exempt(
self, client: httpx.AsyncClient, auth_token: str
):
"""Decky file-drop (multipart/form-data) is NOT rejected with 415."""
resp = await client.post(
"/api/v1/deckies/files/some-container",
files={"file": ("test.txt", b"data", "text/plain")},
headers={"Authorization": f"Bearer {auth_token}"},
)
# Expect 4xx business error (no real container), never 415.
assert resp.status_code != 415
@pytest.mark.asyncio
async def test_empty_body_post_not_enforced(
self, client: httpx.AsyncClient, auth_token: str
):
"""POST with genuinely empty body (Content-Length: 0) is not rejected."""
resp = await client.post(
"/api/v1/logs",
content=b"",
headers={
"Authorization": f"Bearer {auth_token}",
"Content-Length": "0",
},
)
# Middleware should not 415 on empty bodies.
assert resp.status_code != 415
# ---------------------------------------------------------------------------
# BUG-17 — SSE error log uses user["uuid"], not last_event_id
# ---------------------------------------------------------------------------
class TestSSEErrorLog:
def test_sse_error_log_uses_user_uuid(self):
"""
Verify the log.exception call in the SSE generator uses user["uuid"],
not last_event_id (which is an int cursor, not an identity).
"""
import ast, pathlib
src = pathlib.Path(
"decnet/web/router/stream/api_stream_events.py"
).read_text()
tree = ast.parse(src)
bad_pattern_found = False
correct_pattern_found = False
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
# Look for log.exception(...) calls
func = node.func
if not (isinstance(func, ast.Attribute) and func.attr == "exception"):
continue
# Check args after the format string
if len(node.args) >= 2:
arg = node.args[1]
# Bad pattern: bare Name "last_event_id"
if isinstance(arg, ast.Name) and arg.id == "last_event_id":
bad_pattern_found = True
# Good pattern: user["uuid"] subscript
if (
isinstance(arg, ast.Subscript)
and isinstance(arg.value, ast.Name)
and arg.value.id == "user"
):
correct_pattern_found = True
assert not bad_pattern_found, (
"BUG-17: log.exception still uses last_event_id instead of user['uuid']"
)
assert correct_pattern_found, (
"BUG-17 fix not found: expected log.exception(..., user['uuid']) in SSE handler"
)
@pytest.mark.asyncio
async def test_sse_stream_unauthenticated_401(self, client: httpx.AsyncClient):
"""SSE endpoint rejects unauthenticated requests (regression guard)."""
resp = await client.get("/api/v1/stream")
assert resp.status_code == 401

20
tests/core/test_env_secrets.py
View File

@@ -4,7 +4,7 @@
DECNET_ADMIN_PASSWORD must never silently default to "admin": it is resolved
lazily and validated like DECNET_JWT_SECRET. These tests drive _require_env
against a controlled environ so the production raise paths (which are bypassed
under live pytest) are actually exercised.
under the test harness via DECNET_TESTING=1) are actually exercised.
"""
from __future__ import annotations
@@ -14,8 +14,8 @@ import decnet.env as envmod
def _require(monkeypatch: pytest.MonkeyPatch, environ: dict[str, str]) -> str:
# Replace the whole environ for the call so the PYTEST_* short-circuit in
# _require_env doesn't fire — we want the real production behaviour.
# Replace the whole environ for the call so the DECNET_TESTING short-circuit
# in _require_env doesn't fire — we want the real production behaviour.
monkeypatch.setattr(envmod.os, "environ", dict(environ))
return envmod._require_env("DECNET_ADMIN_PASSWORD")
@@ -36,6 +36,20 @@ def test_admin_password_other_known_bad_raises(monkeypatch: pytest.MonkeyPatch,
_require(monkeypatch, {"DECNET_ADMIN_PASSWORD": bad})
def test_known_bad_message_does_not_leak_secret_value(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# V7.1.3: the known-bad rejection must NOT echo the rejected secret value
# (it would land in logs / stderr / crash reporters). Name the variable,
# not its value.
secret = "admin"
with pytest.raises(ValueError) as exc:
_require(monkeypatch, {"DECNET_ADMIN_PASSWORD": secret})
msg = str(exc.value)
assert secret not in msg
assert "DECNET_ADMIN_PASSWORD" in msg
def test_admin_password_too_short_raises_in_production(monkeypatch: pytest.MonkeyPatch) -> None:
with pytest.raises(ValueError, match="too short"):
_require(monkeypatch, {"DECNET_ADMIN_PASSWORD": "short1"})

15
tests/db/mysql/test_mysql_url_builder.py
View File

@@ -13,7 +13,7 @@ def test_build_url_defaults(monkeypatch):
for v in ("DECNET_DB_HOST", "DECNET_DB_PORT", "DECNET_DB_NAME",
"DECNET_DB_USER", "DECNET_DB_PASSWORD", "DECNET_DB_URL"):
monkeypatch.delenv(v, raising=False)
# PYTEST_* is set by pytest itself, so empty password is allowed here.
# DECNET_TESTING=1 is set by conftest, so empty password is allowed here.
url = build_mysql_url()
assert url == "mysql+asyncmy://decnet:@localhost:3306/decnet"
@@ -66,14 +66,13 @@ def test_resolve_url_falls_back_to_components(monkeypatch):
assert url.startswith("mysql+asyncmy://")
def test_build_url_requires_password_outside_pytest(monkeypatch):
"""Without a password and not in a pytest run, construction must fail loudly."""
def test_build_url_requires_password_outside_tests(monkeypatch):
"""Without a password and not under the test harness, construction must fail loudly."""
for v in ("DECNET_DB_URL", "DECNET_DB_PASSWORD"):
monkeypatch.delenv(v, raising=False)
# Strip every PYTEST_* env var so the safety check trips.
import os
for k in list(os.environ):
if k.startswith("PYTEST"):
monkeypatch.delenv(k, raising=False)
# Clear the test-harness flag so the safety check trips. A leaked PYTEST_*
# var must NOT re-enable the bypass (V2.1.7).
monkeypatch.delenv("DECNET_TESTING", raising=False)
monkeypatch.setenv("PYTEST_CURRENT_TEST", "x")
with pytest.raises(ValueError, match="DECNET_DB_PASSWORD is not set"):
build_mysql_url()

12
tests/swarm/test_forwarder_resilience.py
View File

@@ -69,6 +69,18 @@ async def _wait_for(pred, timeout: float = 5.0, interval: float = 0.1) -> bool:
# ----------------------------------------------------------- pure helpers
def test_worker_ssl_context_pins_tls12_floor(_pki_env: dict) -> None:
"""V9.1.4: forwarder client context must set an explicit TLS 1.2 floor."""
ctx = fwd.build_worker_ssl_context(_pki_env["worker_dir"])
assert ctx.minimum_version == ssl.TLSVersion.TLSv1_2
def test_listener_ssl_context_pins_tls12_floor(_pki_env: dict) -> None:
"""V9.1.4: listener server context must set an explicit TLS 1.2 floor."""
ctx = lst.build_listener_ssl_context(_pki_env["ca_dir"])
assert ctx.minimum_version == ssl.TLSVersion.TLSv1_2
def test_peer_cn_returns_unknown_when_no_ssl_object() -> None:
assert lst.peer_cn(None) == "unknown"

66
tests/updater/test_updater_app.py
View File

@@ -54,10 +54,13 @@ def test_health_returns_role_and_releases(client: TestClient, monkeypatch: pytes
def test_update_happy_path(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
ex, "run_update",
lambda data, sha, install_dir, agent_dir, expected_sha256=None: {"status": "updated", "release": {"slot": "active", "sha": sha}, "probe": "ok"},
)
seen: dict[str, object] = {}
def _run_update(data, sha, expected_sha256, install_dir, agent_dir):
seen["expected_sha256"] = expected_sha256
return {"status": "updated", "release": {"slot": "active", "sha": sha}, "probe": "ok"}
monkeypatch.setattr(ex, "run_update", _run_update)
r = client.post(
"/update",
files={"tarball": ("tree.tgz", _tarball(), "application/gzip")},
@@ -65,6 +68,8 @@ def test_update_happy_path(client: TestClient, monkeypatch: pytest.MonkeyPatch)
)
assert r.status_code == 200, r.text
assert r.json()["release"]["sha"] == "ABC123"
# Route forwards the digest verbatim — executor verifies it before extract.
assert seen["expected_sha256"] == "0" * 64
def test_update_rollback_returns_409(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -104,10 +109,13 @@ def test_update_self_requires_confirm(client: TestClient) -> None:
def test_update_self_happy_path(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
ex, "run_update_self",
lambda data, sha, updater_install_dir, expected_sha256=None: {"status": "self_update_queued", "argv": ["python", "-m", "decnet", "updater"]},
)
seen: dict[str, object] = {}
def _run_update_self(data, sha, updater_install_dir, expected_sha256):
seen["expected_sha256"] = expected_sha256
return {"status": "self_update_queued", "argv": ["python", "-m", "decnet", "updater"]}
monkeypatch.setattr(ex, "run_update_self", _run_update_self)
r = client.post(
"/update-self",
files={"tarball": ("t.tgz", _tarball(), "application/gzip")},
@@ -115,6 +123,7 @@ def test_update_self_happy_path(client: TestClient, monkeypatch: pytest.MonkeyPa
)
assert r.status_code == 200
assert r.json()["status"] == "self_update_queued"
assert seen["expected_sha256"] == "0" * 64
def test_rollback_happy(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -158,6 +167,47 @@ def test_update_without_sha256_is_rejected(client: TestClient) -> None:
assert "sha256" in r.json()["detail"]
def test_update_with_empty_sha256_is_rejected(client: TestClient) -> None:
# An explicit empty form value is treated the same as absent → 400.
r = client.post(
"/update",
files={"tarball": ("t.tgz", _tarball(), "application/gzip")},
data={"sha": "ABC", "sha256": ""},
)
assert r.status_code == 400
assert "sha256" in r.json()["detail"]
def test_update_self_without_sha256_is_rejected(client: TestClient) -> None:
r = client.post(
"/update-self",
files={"tarball": ("t.tgz", _tarball(), "application/gzip")},
data={"confirm_self": "true"},
)
assert r.status_code == 400
assert "sha256" in r.json()["detail"]
def test_update_mismatched_sha256_is_rejected_before_apply(
client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
"""End-to-end through the REAL executor verify: a non-matching digest is a
500 UpdateError and no extraction/pip happens (extract/_run_pip would be
reached only AFTER the digest check, so we assert they are never called)."""
called: list[str] = []
monkeypatch.setattr(ex, "extract_tarball", lambda *a, **k: called.append("extract"))
monkeypatch.setattr(ex, "_run_pip", lambda *a, **k: called.append("pip"))
r = client.post(
"/update",
files={"tarball": ("t.tgz", _tarball(), "application/gzip")},
data={"sha": "ABC", "sha256": "0" * 64}, # wrong digest for this tarball
)
assert r.status_code == 500, r.text
assert "mismatch" in r.json()["detail"]["error"]
assert called == [] # rejected before any extract/install
# ------------------------- master-cert gate ---------------------------------

53
tests/updater/test_updater_executor.py
View File

@@ -8,6 +8,7 @@ against a ``tmp_path`` install dir.
"""
from __future__ import annotations
import hashlib
import io
import pathlib
import subprocess
@@ -32,6 +33,11 @@ def _make_tarball(files: dict[str, str]) -> bytes:
return buf.getvalue()
def _digest(tarball: bytes) -> str:
"""SHA-256 hex of the tarball — now mandatory on run_update/run_update_self."""
return hashlib.sha256(tarball).hexdigest()
class _PipOK:
returncode = 0
stdout = ""
@@ -207,6 +213,40 @@ def test_run_update_rejects_malformed_sha256(
)
@pytest.mark.parametrize("missing", ["", " ", None])
def test_run_update_rejects_missing_sha256_fail_closed(
install_dir: pathlib.Path, agent_dir: pathlib.Path, missing: Any,
) -> None:
"""V12.1.2 fail-closed: an absent/empty digest is rejected BEFORE any
extraction or pip-install. No staging tree is produced."""
tb = _make_tarball({"x.txt": "y"})
with pytest.raises(ex.UpdateError, match="required but was missing or empty"):
ex.run_update(
tb, sha="S", expected_sha256=missing, # type: ignore[arg-type]
install_dir=install_dir, agent_dir=agent_dir,
)
assert not (install_dir / "releases" / "active.new").exists()
@pytest.mark.parametrize("missing", ["", " ", None])
def test_run_update_self_rejects_missing_sha256_fail_closed(
install_dir: pathlib.Path, missing: Any,
) -> None:
active = install_dir / "releases" / "active"
active.mkdir()
(active / "marker").write_text("old-updater")
tb = _make_tarball({"marker": "new-updater"})
with pytest.raises(ex.UpdateError, match="required but was missing or empty"):
ex.run_update_self(
tb, sha="U", updater_install_dir=install_dir,
expected_sha256=missing, # type: ignore[arg-type]
exec_cb=lambda a: None,
)
# Active untouched, nothing staged.
assert (install_dir / "releases" / "active" / "marker").read_text() == "old-updater"
assert not (install_dir / "releases" / "active.new").exists()
def test_clean_stale_staging(install_dir: pathlib.Path) -> None:
staging = install_dir / "releases" / "active.new"
staging.mkdir()
@@ -229,7 +269,7 @@ def test_update_rotates_and_probes(
monkeypatch.setattr(ex, "_probe_agent", lambda **_: (True, "ok"))
tb = _make_tarball({"marker.txt": "new"})
result = ex.run_update(tb, sha="NEWSHA", install_dir=install_dir, agent_dir=agent_dir)
result = ex.run_update(tb, sha="NEWSHA", expected_sha256=_digest(tb), install_dir=install_dir, agent_dir=agent_dir)
assert result["status"] == "updated"
assert result["release"]["sha"] == "NEWSHA"
@@ -252,7 +292,7 @@ def test_update_first_install_without_previous(
monkeypatch.setattr(ex, "_probe_agent", lambda **_: (True, "ok"))
tb = _make_tarball({"marker.txt": "first"})
result = ex.run_update(tb, sha="S1", install_dir=install_dir, agent_dir=agent_dir)
result = ex.run_update(tb, sha="S1", expected_sha256=_digest(tb), install_dir=install_dir, agent_dir=agent_dir)
assert result["status"] == "updated"
assert not (install_dir / "releases" / "prev").exists()
@@ -273,7 +313,7 @@ def test_update_pip_failure_aborts_before_rotation(
tb = _make_tarball({"marker.txt": "new"})
with pytest.raises(ex.UpdateError, match="pip install failed") as ei:
ex.run_update(tb, sha="S", install_dir=install_dir, agent_dir=agent_dir)
ex.run_update(tb, sha="S", expected_sha256=_digest(tb), install_dir=install_dir, agent_dir=agent_dir)
assert "resolver error" in ei.value.stderr
# Nothing rotated — old active still live, no prev created.
@@ -309,7 +349,7 @@ def test_update_probe_failure_rolls_back(
tb = _make_tarball({"marker.txt": "new"})
with pytest.raises(ex.UpdateError, match="health probe") as ei:
ex.run_update(tb, sha="NEWSHA", install_dir=install_dir, agent_dir=agent_dir)
ex.run_update(tb, sha="NEWSHA", expected_sha256=_digest(tb), install_dir=install_dir, agent_dir=agent_dir)
assert ei.value.rolled_back is True
assert "connection refused" in ei.value.stderr
@@ -381,6 +421,7 @@ def test_update_self_rotates_and_calls_exec_cb(
tb = _make_tarball({"marker": "new-updater"})
result = ex.run_update_self(
tb, sha="USHA", updater_install_dir=install_dir,
expected_sha256=_digest(tb),
exec_cb=lambda argv: seen_argv.append(argv),
)
assert result["status"] == "self_update_queued"
@@ -412,7 +453,7 @@ def test_update_self_under_systemd_defers_to_systemctl(
monkeypatch.setattr(ex.os, "execv", lambda *a, **k: pytest.fail("execv taken under systemd"))
tb = _make_tarball({"marker": "new-updater"})
result = ex.run_update_self(tb, sha="USHA", updater_install_dir=install_dir)
result = ex.run_update_self(tb, sha="USHA", updater_install_dir=install_dir, expected_sha256=_digest(tb))
assert result == {"status": "self_update_queued", "via": "systemd"}
assert len(popen_calls) == 1
sh_cmd = popen_calls[0]
@@ -431,7 +472,7 @@ def test_update_self_pip_failure_leaves_active_intact(
tb = _make_tarball({"marker": "new-updater"})
with pytest.raises(ex.UpdateError, match="pip install failed"):
ex.run_update_self(tb, sha="U", updater_install_dir=install_dir, exec_cb=lambda a: None)
ex.run_update_self(tb, sha="U", updater_install_dir=install_dir, expected_sha256=_digest(tb), exec_cb=lambda a: None)
assert (install_dir / "releases" / "active" / "marker").read_text() == "old-updater"
assert not (install_dir / "releases" / "active.new").exists()

53
tests/web/test_api_startup_guards.py
View File

@@ -23,13 +23,6 @@ def _reload_api(monkeypatch: pytest.MonkeyPatch):
return importlib.import_module("decnet.web.api")
def _strip_pytest_vars(monkeypatch: pytest.MonkeyPatch) -> None:
import os
for k in list(os.environ):
if k.startswith("PYTEST"):
monkeypatch.delenv(k, raising=False)
async def _run_lifespan_startup(api_mod) -> None:
"""Run the lifespan up to (but not past) yield, then unwind cleanly.
@@ -59,6 +52,7 @@ def test_master_api_refuses_to_start_in_agent_mode(
monkeypatch.setenv("DECNET_MODE", "agent")
monkeypatch.setenv("DECNET_DISALLOW_MASTER", "true")
monkeypatch.setenv("DECNET_JWT_SECRET", "x" * 32)
monkeypatch.setenv("DECNET_CORS_ORIGINS", "http://localhost:8080")
api = _reload_api(monkeypatch)
with pytest.raises(RuntimeError, match="master-only"):
asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
@@ -74,6 +68,7 @@ def test_master_api_starts_when_dual_role_enabled(
monkeypatch.setenv("DECNET_MODE", "agent")
monkeypatch.setenv("DECNET_DISALLOW_MASTER", "false")
monkeypatch.setenv("DECNET_JWT_SECRET", "x" * 32)
monkeypatch.setenv("DECNET_CORS_ORIGINS", "http://localhost:8080")
api = _reload_api(monkeypatch)
# Reaching the DB init phase means the gate passed; we don't need to
# actually finish startup. Cancel via a synthetic exception that the
@@ -107,7 +102,7 @@ def test_master_api_eager_loads_jwt_secret_at_startup(
monkeypatch.setenv("DECNET_MODE", "master")
monkeypatch.setenv("DECNET_JWT_SECRET", "y" * 32)
monkeypatch.setenv("DECNET_API_HOST", "127.0.0.1")
monkeypatch.delenv("DECNET_CORS_ORIGINS", raising=False)
monkeypatch.setenv("DECNET_CORS_ORIGINS", "http://localhost:8080")
api = _reload_api(monkeypatch)
import decnet.env as env_mod
@@ -130,3 +125,45 @@ def test_master_api_eager_loads_jwt_secret_at_startup(
assert "DECNET_JWT_SECRET" in seen, (
"lifespan must access env.DECNET_JWT_SECRET at startup"
)
# ---------------------------------------------------------------------------
# V13.1.4 — CORS wildcard guard
# ---------------------------------------------------------------------------
def test_cors_wildcard_guard_function_raises() -> None:
"""_check_cors_origins raises ValueError when '*' is in the list."""
from decnet.web.api import _check_cors_origins
with pytest.raises(ValueError, match="wildcard"):
_check_cors_origins(["*"])
def test_cors_wildcard_among_explicit_origins_raises() -> None:
"""Wildcard in a mixed list is still rejected."""
from decnet.web.api import _check_cors_origins
with pytest.raises(ValueError, match="wildcard"):
_check_cors_origins(["https://example.com", "*"])
def test_cors_explicit_origins_pass() -> None:
"""Explicit origin URLs pass the guard without raising."""
from decnet.web.api import _check_cors_origins
_check_cors_origins(["https://example.com", "https://app.internal"])
def test_cors_wildcard_raises_in_lifespan(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Lifespan raises ValueError when DECNET_CORS_ORIGINS contains '*'.
Uses _reload_api to pick up the patched env; tests the full guard
path including the lifespan call to _check_cors_origins.
"""
monkeypatch.setenv("DECNET_MODE", "master")
monkeypatch.setenv("DECNET_JWT_SECRET", "z" * 32)
monkeypatch.setenv("DECNET_CORS_ORIGINS", "*")
api = _reload_api(monkeypatch)
with pytest.raises(ValueError, match="wildcard"):
asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
_run_lifespan_startup(api)
)

45
tests/web/test_validate_public_binding.py
View File

@@ -1,9 +1,9 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""validate_public_binding refuses footgun configs at master startup.
The validator no-ops under pytest by design (so unit tests in unrelated
modules don't have to set five env vars per fixture); these tests strip
the PYTEST_* vars before calling it so the real code path runs.
The validator no-ops under the test harness (DECNET_TESTING=1) by design (so
unit tests in unrelated modules don't have to set five env vars per fixture);
these tests clear that flag before calling it so the real code path runs.
"""
from __future__ import annotations
@@ -20,18 +20,17 @@ def _reimport_env(monkeypatch: pytest.MonkeyPatch):
return importlib.import_module("decnet.env")
def _strip_pytest_vars(monkeypatch: pytest.MonkeyPatch) -> None:
import os
for k in list(os.environ):
if k.startswith("PYTEST"):
monkeypatch.delenv(k, raising=False)
def _strip_test_flag(monkeypatch: pytest.MonkeyPatch) -> None:
# The validator short-circuits on DECNET_TESTING=1 (set globally by
# conftest). Clear it so the real production code path runs.
monkeypatch.delenv("DECNET_TESTING", raising=False)
def test_validator_noop_on_loopback_binding(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("DECNET_API_HOST", "127.0.0.1")
monkeypatch.setenv("DECNET_CORS_ORIGINS", "http://localhost:8080")
env = _reimport_env(monkeypatch)
_strip_pytest_vars(monkeypatch)
_strip_test_flag(monkeypatch)
env.validate_public_binding() # no raise
@@ -41,7 +40,7 @@ def test_validator_rejects_loopback_cors_on_public_bind(
monkeypatch.setenv("DECNET_API_HOST", "0.0.0.0")
monkeypatch.setenv("DECNET_CORS_ORIGINS", "http://localhost:8080")
env = _reimport_env(monkeypatch)
_strip_pytest_vars(monkeypatch)
_strip_test_flag(monkeypatch)
with pytest.raises(ValueError, match="loopback origin"):
env.validate_public_binding()
@@ -52,7 +51,7 @@ def test_validator_accepts_public_cors_on_public_bind(
monkeypatch.setenv("DECNET_API_HOST", "0.0.0.0")
monkeypatch.setenv("DECNET_CORS_ORIGINS", "https://dashboard.example.com")
env = _reimport_env(monkeypatch)
_strip_pytest_vars(monkeypatch)
_strip_test_flag(monkeypatch)
env.validate_public_binding() # no raise
@@ -63,7 +62,7 @@ def test_validator_rejects_plaintext_canary_on_public_bind(
monkeypatch.setenv("DECNET_CORS_ORIGINS", "https://dashboard.example.com")
monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "http://canary.example.com:8088")
env = _reimport_env(monkeypatch)
_strip_pytest_vars(monkeypatch)
_strip_test_flag(monkeypatch)
with pytest.raises(ValueError, match="plaintext HTTP"):
env.validate_public_binding()
@@ -77,14 +76,28 @@ def test_validator_allows_loopback_canary_even_on_public_bind(
monkeypatch.setenv("DECNET_CORS_ORIGINS", "https://dashboard.example.com")
monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "http://localhost:8088")
env = _reimport_env(monkeypatch)
_strip_pytest_vars(monkeypatch)
_strip_test_flag(monkeypatch)
env.validate_public_binding() # no raise
def test_validator_skips_under_pytest(monkeypatch: pytest.MonkeyPatch) -> None:
# With PYTEST_* still in env (default), even a misconfigured env passes —
# this is the deliberate bypass so unrelated tests don't trip on it.
def test_validator_skips_under_test_harness(monkeypatch: pytest.MonkeyPatch) -> None:
# With DECNET_TESTING=1 still in env (set by conftest), even a misconfigured
# env passes — this is the deliberate bypass so unrelated tests don't trip.
monkeypatch.setenv("DECNET_TESTING", "1")
monkeypatch.setenv("DECNET_API_HOST", "0.0.0.0")
monkeypatch.setenv("DECNET_CORS_ORIGINS", "http://localhost:8080")
env = _reimport_env(monkeypatch)
env.validate_public_binding() # no raise — guard short-circuits
def test_validator_pytest_var_leak_does_not_bypass(monkeypatch: pytest.MonkeyPatch) -> None:
# V2.1.7 regression: a leaked PYTEST_* env var must NOT disable the guard.
# With DECNET_TESTING cleared, a misconfigured public binding still raises
# even though a PYTEST_* var is present.
monkeypatch.delenv("DECNET_TESTING", raising=False)
monkeypatch.setenv("PYTEST_CURRENT_TEST", "x")
monkeypatch.setenv("DECNET_API_HOST", "0.0.0.0")
monkeypatch.setenv("DECNET_CORS_ORIGINS", "http://localhost:8080")
env = _reimport_env(monkeypatch)
with pytest.raises(ValueError, match="loopback origin"):
env.validate_public_binding()