Spaces:
Configuration error
Configuration error
File size: 5,106 Bytes
fce12d3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """
Auth matrix — every registered HTTP rule in the Flask app.
For each rule this test asserts:
1. 401 when a secret is set and no token is provided
2. non-401 when a valid token is provided
3. non-401 when no secret is configured (auth disabled mode)
Rules explicitly marked in ``UNPROTECTED_PATHS`` are expected to skip auth
even when the secret is set (livez, health, auth.md, static viewer routes).
Rules that don't take our synthetic empty payload cleanly (e.g. WebSocket,
POST endpoints requiring specific fields) may 4xx/5xx on the *body* — that's
still not a 401, which is all this matrix cares about.
"""
from __future__ import annotations
import pytest
import agentcache.app as app_mod
from agentcache.app import create_app
UNPROTECTED_PATHS = {
"/auth.md",
"/agentcache/livez",
"/agentmemory/livez",
"/agentcache/health",
"/agentmemory/health",
"/",
"/viewer",
"/agentcache/viewer",
"/agentmemory/viewer",
"/favicon.svg",
}
def _reset_app_globals() -> None:
app_mod.kv = None
app_mod.search_service = None
app_mod.observation_store = None
def _iter_http_rules(app):
"""Yield (method, path) for every non-static HTTP rule, one method per row."""
for rule in app.url_map.iter_rules():
# skip rules with URL parameters — we can't synthesise a valid one
if "<" in rule.rule:
continue
# skip static asset endpoint
if rule.endpoint == "static":
continue
# skip WebSocket rules — they can't be exercised via HTTP test client
if rule.rule.startswith("/stream/"):
continue
methods = (rule.methods or set()) - {"HEAD", "OPTIONS"}
for method in sorted(methods):
yield method, rule.rule
@pytest.fixture
def secured_app(tmp_path, monkeypatch):
monkeypatch.setenv("AGENTCACHE_DB_PATH", str(tmp_path / "matrix_secured.db"))
monkeypatch.setenv("AGENTCACHE_SECRET", "matrix-secret")
_reset_app_globals()
return create_app()
@pytest.fixture
def open_app(tmp_path, monkeypatch):
monkeypatch.setenv("AGENTCACHE_DB_PATH", str(tmp_path / "matrix_open.db"))
monkeypatch.delenv("AGENTCACHE_SECRET", raising=False)
monkeypatch.delenv("AGENTMEMORY_SECRET", raising=False)
_reset_app_globals()
return create_app()
def _invoke(client, method: str, path: str, headers=None):
fn = getattr(client, method.lower())
if method in ("POST", "PUT", "PATCH", "DELETE"):
return fn(path, json={}, headers=headers or {})
return fn(path, headers=headers or {})
def test_every_route_requires_auth_when_secret_is_set(secured_app):
"""Protected routes 401 without a token; unprotected routes pass through."""
client = secured_app.test_client()
failures = []
for method, path in _iter_http_rules(secured_app):
res = _invoke(client, method, path)
if path in UNPROTECTED_PATHS:
if res.status_code == 401:
failures.append(
f"{method} {path} unexpectedly required auth (401)"
)
else:
if res.status_code != 401:
failures.append(
f"{method} {path} did not require auth (got {res.status_code})"
)
assert not failures, "auth-matrix mismatches:\n " + "\n ".join(failures)
def test_every_route_accepts_valid_token(secured_app):
"""With a valid Bearer token, no route returns 401."""
client = secured_app.test_client()
headers = {"Authorization": "Bearer matrix-secret"}
failures = []
for method, path in _iter_http_rules(secured_app):
res = _invoke(client, method, path, headers=headers)
if res.status_code == 401:
failures.append(f"{method} {path} rejected a valid token (401)")
assert not failures, "rejected valid token:\n " + "\n ".join(failures)
def test_every_route_open_when_no_secret_configured(open_app):
"""When no secret env var is set, every route passes auth (no 401 anywhere)."""
client = open_app.test_client()
failures = []
for method, path in _iter_http_rules(open_app):
res = _invoke(client, method, path)
if res.status_code == 401:
failures.append(
f"{method} {path} returned 401 in open (no-secret) mode"
)
assert not failures, "unexpected 401s in open mode:\n " + "\n ".join(failures)
def test_every_route_rejects_invalid_token(secured_app):
"""With a wrong Bearer token, protected routes must still 401."""
client = secured_app.test_client()
bad_headers = {"Authorization": "Bearer not-the-secret"}
failures = []
for method, path in _iter_http_rules(secured_app):
if path in UNPROTECTED_PATHS:
continue
res = _invoke(client, method, path, headers=bad_headers)
if res.status_code != 401:
failures.append(
f"{method} {path} accepted a bad token (got {res.status_code})"
)
assert not failures, "invalid-token acceptance:\n " + "\n ".join(failures)
|