143 lines
6.6 KiB
Python
143 lines
6.6 KiB
Python
"""Härtungen aus dem Security-Review (→ KONZEPT.md 16)."""
|
|
|
|
import io
|
|
import zipfile
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from hugin_app.app import create_app
|
|
from hugin_app.rate_limit import Window
|
|
from hugin_app.settings import Settings
|
|
from hugin_app.hugin import Hugin
|
|
from hugin_app.migration import migrate
|
|
from test_app import ADMIN, tenant_package, env, zip_from_files # noqa: F401
|
|
|
|
ZIP = {"Content-Type": "application/zip"}
|
|
MANIFEST = "id: shop-x\nkind: rules\nversion: 1.0.0\nowner: api\nscope: tenant\ndescription: x\n"
|
|
|
|
|
|
@pytest.fixture
|
|
def small(database_url) -> TestClient:
|
|
migrate(database_url)
|
|
hugin = Hugin(Settings(database_url=database_url, admin_token="admin-geheim", max_upload_mb=1))
|
|
return TestClient(create_app(hugin))
|
|
|
|
|
|
def test_large_bodies_rejected_before_login(small):
|
|
large = b"\0" * ((1 << 20) + 1)
|
|
# mit Content-Length: sofort, auch ohne Key
|
|
assert small.post("/api/v1/validate", content=large, headers=ZIP).status_code == 413
|
|
# ohne Content-Length (chunked): sobald die Grenze erreicht ist
|
|
read = 0
|
|
|
|
def chunked():
|
|
nonlocal read
|
|
for _ in range(8):
|
|
read += 256 << 10
|
|
yield b"\0" * (256 << 10)
|
|
|
|
assert small.post("/api/v1/validate", content=chunked(), headers=ZIP).status_code == 413
|
|
# kleine Bodies gehen wie gewohnt bis zur Anmeldung
|
|
assert small.post("/api/v1/validate", content=b"x", headers=ZIP).status_code == 401
|
|
|
|
|
|
def test_no_pages_in_browser_cache(env):
|
|
assert env.client.get("/help").headers["cache-control"] == "no-store"
|
|
assert env.client.get("/api/v1/packages", headers=env.h("shop-lesen")).headers["cache-control"] == "no-store"
|
|
assert "no-store" not in env.client.get("/static/css/hugin.css").headers.get("cache-control", "")
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["a:b.md", "nul.md", "CON", "x/file.", "datei ", "a|b", "frage?.md"])
|
|
def test_filenames_for_all_systems(env, name):
|
|
raw = zip_from_files({"hugin.yaml": MANIFEST, "RULES.md": "x\n", name: "x"})
|
|
response = env.client.post("/api/v1/validate", content=raw, headers=env.h("shop-autor") | ZIP)
|
|
assert response.status_code == 422 and "nicht auf allen Systemen" in response.text
|
|
|
|
|
|
def test_broken_zips_without_server_error(env):
|
|
raw = bytearray(zip_from_files({"hugin.yaml": MANIFEST, "RULES.md": "x" * 200}))
|
|
position = raw.find(b"xxxx")
|
|
raw[position:position + 4] = b"yyyy" # Prüfsumme stimmt nicht mehr
|
|
not_utf8 = zip_from_files({"hugin.yaml": "x"}).replace(b"x", b"\xff", 1)
|
|
buffer = io.BytesIO()
|
|
with zipfile.ZipFile(buffer, "w") as z:
|
|
z.writestr("hugin.yaml", b"id: \xff\n")
|
|
for body in (bytes(raw), buffer.getvalue(), not_utf8, b"PK\x03\x04kaputt"):
|
|
response = env.client.post("/api/v1/validate", content=body, headers=env.h("shop-autor") | ZIP)
|
|
assert response.status_code == 422, response.text
|
|
|
|
|
|
def test_too_many_files(env):
|
|
files = {"hugin.yaml": MANIFEST, "RULES.md": "x"} | {f"d/{i}.md": "x" for i in range(501)}
|
|
response = env.client.post("/api/v1/validate", content=zip_from_files(files),
|
|
headers=env.h("shop-autor") | ZIP)
|
|
assert response.status_code == 422 and "500 Dateien" in response.text
|
|
|
|
|
|
def test_linter_reveals_no_foreign_versions(env):
|
|
tenant_package(env) # shop-intern 1.0.0, nur im Tenant shop sichtbar
|
|
manifest = MANIFEST.replace("shop-x", "shop-intern").replace("owner: api", "owner: crm")
|
|
raw = zip_from_files({"hugin.yaml": manifest, "RULES.md": "x\n"})
|
|
foreign = env.client.post("/api/v1/validate", content=raw, headers=env.h("vertrieb-lesen") | ZIP).json()
|
|
assert "1.0.0 muss größer" not in str(foreign["findings"])
|
|
own = env.client.post("/api/v1/validate", content=raw, headers=env.h("shop-lesen") | ZIP).json()
|
|
assert "muss größer sein als 1.0.0" in str(own["findings"])
|
|
|
|
|
|
def test_rate_limit_memory_stays_limited(monkeypatch):
|
|
now = [1000.0]
|
|
monkeypatch.setattr("hugin_app.rate_limit.time.monotonic", lambda: now[0])
|
|
f = Window(5, 60)
|
|
for i in range(12_000):
|
|
f.allow(f"zufall-{i}")
|
|
now[0] += 61 # alle abgelaufen
|
|
f.allow("new")
|
|
assert len(f._events) == 1
|
|
# innerhalb eines Fensters wird nicht bei jeder Anfrage neu aufgeräumt
|
|
for i in range(12_000):
|
|
f.allow(f"mehr-{i}")
|
|
assert len(f._events) == 12_001
|
|
|
|
|
|
def test_proxy_only_trusted(monkeypatch):
|
|
from uvicorn.middleware.proxy_headers import _TrustedHosts
|
|
|
|
monkeypatch.delenv("HUGIN_PROXY_IPS", raising=False)
|
|
default = Settings.from_env().proxy_ips
|
|
assert "1.2.3.4" not in _TrustedHosts(default) and "10.0.0.5" not in _TrustedHosts(default)
|
|
monkeypatch.setenv("HUGIN_PROXY_IPS", "10.128.0.0/14")
|
|
assert "10.128.3.7" in _TrustedHosts(Settings.from_env().proxy_ips)
|
|
|
|
|
|
def test_integrations_key_only_for_delivery(env):
|
|
env.example("pub-hausregeln/1.0.0")
|
|
env.key("shop", "shop-integration", kind="integration")
|
|
h = env.h("shop-integration")
|
|
c = env.client
|
|
assert c.get("/api/v1/delivery/status", headers=h).status_code == 200
|
|
for response in (c.get("/api/v1/packages", headers=h),
|
|
c.post("/api/v1/bundle", headers=h, json={"packages": ["pub-hausregeln"], "tools": {"opencode": "1.19.3"}}),
|
|
c.get("/api/v1/admin/settings", headers=h)):
|
|
assert response.status_code == 403 and "nur für /api/v1/delivery" in response.text
|
|
mcp = c.post("/mcp", headers=h | {"Accept": "application/json, text/event-stream"},
|
|
json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
|
|
assert mcp.status_code == 401
|
|
web = TestClient(c.app, follow_redirects=False)
|
|
response = web.post("/login", data={"key": env.keys["shop-integration"], "next": ""})
|
|
assert response.status_code == 403 and "hugin_session" not in response.headers.get("set-cookie", "")
|
|
|
|
|
|
def test_admin_session_ends_with_new_token(database_url):
|
|
migrate(database_url)
|
|
old = Hugin(Settings(database_url=database_url, admin_token="alt-geheim"))
|
|
web = TestClient(create_app(old), follow_redirects=False)
|
|
assert web.post("/login", data={"key": "alt-geheim", "next": ""}).status_code == 303
|
|
assert web.get("/manage").status_code == 200
|
|
# Neustart mit gewechseltem Token, gleiche Datenbank, gleiches Cookie
|
|
new = TestClient(create_app(Hugin(Settings(database_url=database_url, admin_token="neu-geheim"))),
|
|
follow_redirects=False, cookies=web.cookies)
|
|
assert new.get("/manage").status_code == 303 # zurück zur Anmeldung
|
|
without = TestClient(create_app(Hugin(Settings(database_url=database_url, admin_token=""))),
|
|
follow_redirects=False, cookies=web.cookies)
|
|
assert without.get("/manage").status_code == 303
|