fix(securitate): hardening prod — headere, body-cap, non-root, backup
Findings security-review P1/P2 (2026-07-03): - headere de securitate pe toate raspunsurile (nosniff, X-Frame-Options, Referrer-Policy, HSTS doar pe HTTPS) + teste - body-cap global 10MB ca middleware ASGI pur (413 inainte de parserul multipart/JSON; verificarea per-endpoint ramane strat 2) - imagine Docker non-root (uid 10001), port 8010 aliniat, loguri pe volumul /data - fail-fast la boot cu rar_env=prod fara AUTOPASS_REQUIRE_API_KEY sau AUTOPASS_SESSION_SECRET - compose: env-uri critice obligatorii (:?) ca api/worker sa nu diverga tacit; FORWARDED_ALLOW_IPS ca rate-limit-ul sa vada IP-ul real dupa Traefik - signup fara PII in stdout: log_event in loc de print cu email (idem notify degradat) - ratelimit: sterge cheile fara timestamp-uri valide (crestere monotona a memoriei pe IP-uri reale) - backup criptat SQLite (backup online API, gpg AES256) + verificare restore + docs/backup.md Suita completa verde: 1557 passed, 1 skipped (live). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
64
tests/test_body_cap.py
Normal file
64
tests/test_body_cap.py
Normal file
@@ -0,0 +1,64 @@
|
||||
"""Teste P1-5 (hardening 2026-07-03): plafon global corp cerere (BodyCapMiddleware)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def client(monkeypatch):
|
||||
tmp = tempfile.mkdtemp()
|
||||
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "bc.db"))
|
||||
monkeypatch.setenv("AUTOPASS_LOG_DIR", os.path.join(tmp, "logs"))
|
||||
# Prag mic ca sa nu generam MB-uri reale in test.
|
||||
monkeypatch.setenv("AUTOPASS_MAX_REQUEST_BYTES", "1000")
|
||||
from app.config import get_settings
|
||||
get_settings.cache_clear()
|
||||
from app.main import app
|
||||
with TestClient(app) as c:
|
||||
yield c
|
||||
get_settings.cache_clear()
|
||||
|
||||
|
||||
def test_multipart_peste_limita_413(client):
|
||||
data = b"x" * 5000
|
||||
r = client.post(
|
||||
"/v1/import",
|
||||
files={"file": ("mare.csv", io.BytesIO(data), "text/csv")},
|
||||
)
|
||||
assert r.status_code == 413
|
||||
assert r.json()["cod"] == "CERERE_PREA_MARE"
|
||||
|
||||
|
||||
def test_json_peste_limita_413(client):
|
||||
payload = {"blob": "x" * 5000}
|
||||
r = client.post(
|
||||
"/v1/prezentari",
|
||||
content=json.dumps(payload),
|
||||
headers={"content-type": "application/json"},
|
||||
)
|
||||
assert r.status_code == 413
|
||||
assert r.json()["cod"] == "CERERE_PREA_MARE"
|
||||
|
||||
|
||||
def test_cerere_normala_sub_limita_trece(client):
|
||||
r = client.get("/healthz")
|
||||
assert r.status_code == 200
|
||||
|
||||
|
||||
def test_content_length_mare_body_mic_413_devreme(client):
|
||||
"""Content-Length declarat peste limita respinge inainte sa citeasca body-ul,
|
||||
chiar daca body-ul trimis efectiv e mic (client "mincinos")."""
|
||||
r = client.post(
|
||||
"/v1/prezentari",
|
||||
content=b"{}",
|
||||
headers={"content-type": "application/json", "content-length": "999999"},
|
||||
)
|
||||
assert r.status_code == 413
|
||||
assert r.json()["cod"] == "CERERE_PREA_MARE"
|
||||
@@ -39,6 +39,9 @@ def client_prod(monkeypatch):
|
||||
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
|
||||
monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "true")
|
||||
monkeypatch.setenv("AUTOPASS_RAR_ENV", "prod")
|
||||
# rar_env=prod declanseaza invarianta de startup (validate_prod_invariants,
|
||||
# hardening 2026-07-03): cere si session_secret setat, altfel boot-ul refuza.
|
||||
monkeypatch.setenv("AUTOPASS_SESSION_SECRET", "x" * 32)
|
||||
from app.config import get_settings
|
||||
get_settings.cache_clear()
|
||||
from app.main import app
|
||||
|
||||
77
tests/test_ratelimit_cleanup.py
Normal file
77
tests/test_ratelimit_cleanup.py
Normal file
@@ -0,0 +1,77 @@
|
||||
"""Teste F5: `_hits` nu creste monoton pe viata procesului.
|
||||
|
||||
Cheile ale caror timestamp-uri au expirat complet trebuie sterse din `_hits`
|
||||
(nu doar golite), altfel dictionarul acumuleaza o intrare per IP vazut vreodata.
|
||||
Semantica ferestrei glisante (permite/blocheaza) trebuie sa ramana neschimbata.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.web import ratelimit
|
||||
from app.web.ratelimit import check_rate_limit
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_hits():
|
||||
ratelimit._hits.clear()
|
||||
yield
|
||||
ratelimit._hits.clear()
|
||||
|
||||
|
||||
def test_cheie_blocata_fara_timestamp_uri_valide_e_stearsa(monkeypatch):
|
||||
"""max_hits=0 -> orice cerere e blocata; daca fereastra a expirat complet
|
||||
(filtrarea produce o lista goala), cheia nu trebuie sa ramana in `_hits`."""
|
||||
now = [1000.0]
|
||||
monkeypatch.setattr(ratelimit.time, "monotonic", lambda: now[0])
|
||||
|
||||
key = "ip_blocat_gol"
|
||||
assert check_rate_limit(key, max_hits=0, window_s=60) is False
|
||||
# Nimic adaugat (max_hits=0 blocheaza tot) si lista era goala -> cheia disparuta.
|
||||
assert key not in ratelimit._hits
|
||||
|
||||
|
||||
def test_cheie_activa_ramane_dupa_permis(monkeypatch):
|
||||
"""O cerere permisa isi adauga timestamp-ul -> cheia activa ramane in `_hits`."""
|
||||
now = [1000.0]
|
||||
monkeypatch.setattr(ratelimit.time, "monotonic", lambda: now[0])
|
||||
|
||||
key = "ip_activ"
|
||||
assert check_rate_limit(key, max_hits=2, window_s=60) is True
|
||||
assert key in ratelimit._hits
|
||||
assert ratelimit._hits[key] == [1000.0]
|
||||
|
||||
|
||||
def test_semantica_max_hits_si_fereastra_neschimbata(monkeypatch):
|
||||
"""max_hits atinse -> False; dupa expirarea ferestrei -> True din nou."""
|
||||
now = [1000.0]
|
||||
monkeypatch.setattr(ratelimit.time, "monotonic", lambda: now[0])
|
||||
|
||||
key = "ip_fereastra"
|
||||
assert check_rate_limit(key, max_hits=2, window_s=10) is True
|
||||
assert check_rate_limit(key, max_hits=2, window_s=10) is True
|
||||
# A treia cerere in aceeasi fereastra -> blocata.
|
||||
assert check_rate_limit(key, max_hits=2, window_s=10) is False
|
||||
|
||||
# Trece timpul peste fereastra -> toate timestamp-urile expira.
|
||||
now[0] += 11
|
||||
assert check_rate_limit(key, max_hits=2, window_s=10) is True
|
||||
# Doar noua cerere ramane in lista (cele vechi, expirate, au fost curatate).
|
||||
assert ratelimit._hits[key] == [1011.0]
|
||||
|
||||
|
||||
def test_cheie_dispare_dupa_expirare_completa_si_reblocare(monkeypatch):
|
||||
"""O cheie blocata, ramasa fara timestamp-uri valide dupa expirarea ferestrei
|
||||
(fara sa mai apara alte cereri intre timp), nu trebuie sa ramana in `_hits`
|
||||
la urmatorul apel care o gaseste goala si tot o blocheaza (max_hits=0)."""
|
||||
now = [1000.0]
|
||||
monkeypatch.setattr(ratelimit.time, "monotonic", lambda: now[0])
|
||||
|
||||
key = "ip_reblocat"
|
||||
check_rate_limit(key, max_hits=0, window_s=5)
|
||||
assert key not in ratelimit._hits
|
||||
|
||||
now[0] += 100
|
||||
assert check_rate_limit(key, max_hits=0, window_s=5) is False
|
||||
assert key not in ratelimit._hits
|
||||
64
tests/test_security_headers.py
Normal file
64
tests/test_security_headers.py
Normal file
@@ -0,0 +1,64 @@
|
||||
"""Teste P1-4 (hardening 2026-07-03): security headers pe TOATE raspunsurile."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def client(monkeypatch):
|
||||
tmp = tempfile.mkdtemp()
|
||||
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "sh.db"))
|
||||
monkeypatch.setenv("AUTOPASS_LOG_DIR", os.path.join(tmp, "logs"))
|
||||
from app.config import get_settings
|
||||
get_settings.cache_clear()
|
||||
from app.main import app
|
||||
with TestClient(app) as c:
|
||||
yield c
|
||||
get_settings.cache_clear()
|
||||
|
||||
|
||||
def _assert_common_headers(headers) -> None:
|
||||
assert headers.get("X-Content-Type-Options") == "nosniff"
|
||||
assert headers.get("X-Frame-Options") == "DENY"
|
||||
assert headers.get("Referrer-Policy") == "strict-origin-when-cross-origin"
|
||||
|
||||
|
||||
def test_headere_pe_raspuns_200(client):
|
||||
r = client.get("/healthz")
|
||||
assert r.status_code == 200
|
||||
_assert_common_headers(r.headers)
|
||||
|
||||
|
||||
def test_headere_pe_raspuns_404(client):
|
||||
r = client.get("/o/ruta/care/nu/exista")
|
||||
assert r.status_code == 404
|
||||
_assert_common_headers(r.headers)
|
||||
|
||||
|
||||
def test_hsts_absent_pe_http(client):
|
||||
r = client.get("/healthz")
|
||||
assert "Strict-Transport-Security" not in r.headers
|
||||
|
||||
|
||||
def test_hsts_prezent_pe_https():
|
||||
tmp = tempfile.mkdtemp()
|
||||
import os as _os
|
||||
|
||||
_os.environ["AUTOPASS_DB_PATH"] = _os.path.join(tmp, "sh_https.db")
|
||||
from app.config import get_settings
|
||||
get_settings.cache_clear()
|
||||
from app.main import app
|
||||
|
||||
with TestClient(app, base_url="https://testserver") as c:
|
||||
r = c.get("/healthz")
|
||||
assert r.status_code == 200
|
||||
assert r.headers.get("Strict-Transport-Security") == (
|
||||
"max-age=31536000; includeSubDomains"
|
||||
)
|
||||
get_settings.cache_clear()
|
||||
_os.environ.pop("AUTOPASS_DB_PATH", None)
|
||||
@@ -155,16 +155,33 @@ def test_primul_signup_devine_admin(client):
|
||||
# Teste C16 (log SIGNUP pastrat) si best-effort E2E
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_signup_inca_logheaza_si_notifica(client, capsys):
|
||||
"""Signup reusit -> stdout contine 'SIGNUP cont=' (C16 pastrat)."""
|
||||
def test_signup_logheaza_fara_pii_in_stdout(client, capsys):
|
||||
"""Signup reusit -> NU se mai printeaza email-ul in clar in stdout (P2-7).
|
||||
|
||||
C16 (jurnalizarea signup-ului) e pastrata, dar prin `log_event` (app_events +
|
||||
log text redactat), nu prin print() cu PII."""
|
||||
resp = _do_signup(client, "Service Log Test", "log@test.com")
|
||||
assert resp.status_code == 200
|
||||
assert "rfak_" in resp.text
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert "SIGNUP cont=" in captured.out, (
|
||||
f"Linia de log C16 'SIGNUP cont=' lipseste din stdout. Capturat: {captured.out!r}"
|
||||
assert "log@test.com" not in captured.out, (
|
||||
f"Emailul nu trebuie sa apara in clar in stdout. Capturat: {captured.out!r}"
|
||||
)
|
||||
assert "SIGNUP cont=" not in captured.out, (
|
||||
"Print-ul vechi cu PII nu mai trebuie sa existe."
|
||||
)
|
||||
|
||||
from app.db import get_connection
|
||||
conn = get_connection()
|
||||
try:
|
||||
row = conn.execute(
|
||||
"SELECT account_id, tip FROM app_events WHERE tip='signup' ORDER BY id DESC LIMIT 1"
|
||||
).fetchone()
|
||||
finally:
|
||||
conn.close()
|
||||
assert row is not None, "Trebuie sa existe un eveniment 'signup' in app_events"
|
||||
assert row["tip"] == "signup"
|
||||
|
||||
|
||||
def test_signup_neblocat_de_notify(monkeypatch, client):
|
||||
|
||||
86
tests/test_startup_invariant.py
Normal file
86
tests/test_startup_invariant.py
Normal file
@@ -0,0 +1,86 @@
|
||||
"""Teste E1 (hardening 2026-07-03): invarianta de startup fail-fast, prod-only.
|
||||
|
||||
`validate_prod_invariants` (app/config.py) trebuie sa refuze boot-ul cand
|
||||
`rar_env=="prod"` si lipseste `require_api_key` sau `session_secret` — dar sa
|
||||
ramana un no-op pe `rar_env` implicit ("test"), altfel ar opri toata suita.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
def _clear(monkeypatch):
|
||||
# NU delenv REQUIRE_API_KEY: `.env` real (dev) are AUTOPASS_REQUIRE_API_KEY=true,
|
||||
# iar conftest il neutralizeaza cu os.environ.setdefault("...", "false") — un delenv
|
||||
# aici ar re-expune valoarea din `.env` (precedenta pydantic-settings: env var > .env
|
||||
# file). Fiecare test seteaza explicit ce valoare vrea pentru acest flag.
|
||||
monkeypatch.delenv("AUTOPASS_RAR_ENV", raising=False)
|
||||
monkeypatch.delenv("AUTOPASS_SESSION_SECRET", raising=False)
|
||||
|
||||
|
||||
def test_prod_fara_require_api_key_refuza_boot(monkeypatch):
|
||||
_clear(monkeypatch)
|
||||
monkeypatch.setenv("AUTOPASS_RAR_ENV", "prod")
|
||||
monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "false")
|
||||
monkeypatch.setenv("AUTOPASS_SESSION_SECRET", "x" * 32)
|
||||
from app.config import get_settings
|
||||
get_settings.cache_clear()
|
||||
from app.main import app
|
||||
|
||||
with pytest.raises(RuntimeError, match="AUTOPASS_REQUIRE_API_KEY"):
|
||||
with TestClient(app):
|
||||
pass
|
||||
get_settings.cache_clear()
|
||||
|
||||
|
||||
def test_prod_fara_session_secret_refuza_boot(monkeypatch):
|
||||
_clear(monkeypatch)
|
||||
monkeypatch.setenv("AUTOPASS_RAR_ENV", "prod")
|
||||
monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "true")
|
||||
from app.config import get_settings
|
||||
get_settings.cache_clear()
|
||||
from app.main import app
|
||||
|
||||
with pytest.raises(RuntimeError, match="AUTOPASS_SESSION_SECRET"):
|
||||
with TestClient(app):
|
||||
pass
|
||||
get_settings.cache_clear()
|
||||
|
||||
|
||||
def test_prod_cu_invarianta_satisfacuta_boot_ok(monkeypatch):
|
||||
_clear(monkeypatch)
|
||||
tmp = tempfile.mkdtemp()
|
||||
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "inv.db"))
|
||||
monkeypatch.setenv("AUTOPASS_LOG_DIR", os.path.join(tmp, "logs"))
|
||||
monkeypatch.setenv("AUTOPASS_RAR_ENV", "prod")
|
||||
monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "true")
|
||||
monkeypatch.setenv("AUTOPASS_SESSION_SECRET", "x" * 32)
|
||||
from app.config import get_settings
|
||||
get_settings.cache_clear()
|
||||
from app.main import app
|
||||
|
||||
with TestClient(app) as c:
|
||||
r = c.get("/healthz")
|
||||
assert r.status_code == 200
|
||||
get_settings.cache_clear()
|
||||
|
||||
|
||||
def test_mediu_test_nu_declanseaza_invarianta(monkeypatch):
|
||||
"""Default rar_env="test" -> no-op, indiferent de require_api_key/session_secret."""
|
||||
_clear(monkeypatch)
|
||||
tmp = tempfile.mkdtemp()
|
||||
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "inv2.db"))
|
||||
monkeypatch.setenv("AUTOPASS_LOG_DIR", os.path.join(tmp, "logs"))
|
||||
from app.config import get_settings
|
||||
get_settings.cache_clear()
|
||||
from app.main import app
|
||||
|
||||
with TestClient(app) as c:
|
||||
r = c.get("/healthz")
|
||||
assert r.status_code == 200
|
||||
get_settings.cache_clear()
|
||||
Reference in New Issue
Block a user