164f90d603
- N1: test HSTS pe monkeypatch.setenv (fara env var scursa la assert picat) - N2: backup_db.sh refuza AUTOPASS_BACKUP_KEEP < 1 (retentia ar fi sters backup-ul abia creat) - N3: teste ASGI directe pentru Content-Length malformat/negativ in BodyCapMiddleware (comportamentul defensiv exista deja, acum e fixat in teste) Suita completa: 1559 passed, 1 skipped. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
132 lines
3.9 KiB
Python
132 lines
3.9 KiB
Python
"""Teste P1-5 (hardening 2026-07-03): plafon global corp cerere (BodyCapMiddleware)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import io
|
|
import json
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.web.body_cap import BodyCapMiddleware
|
|
|
|
|
|
@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"
|
|
|
|
|
|
async def _dummy_app(scope, receive, send):
|
|
"""App ASGI minimal: consuma tot body-ul (ca un endpoint real), apoi 200."""
|
|
more_body = True
|
|
while more_body:
|
|
message = await receive()
|
|
more_body = message.get("more_body", False)
|
|
await send({"type": "http.response.start", "status": 200, "headers": []})
|
|
await send({"type": "http.response.body", "body": b"ok"})
|
|
|
|
|
|
def _run_body_cap(headers, body_chunks):
|
|
"""Ruleaza BodyCapMiddleware direct pe un scope/receive/send ASGI construit manual
|
|
(TestClient/httpx nu lasa Content-Length malformat/negativ sa treaca la request real)."""
|
|
scope = {"type": "http", "method": "POST", "path": "/x", "headers": headers}
|
|
chunks = list(body_chunks)
|
|
|
|
async def receive():
|
|
if not chunks:
|
|
return {"type": "http.request", "body": b"", "more_body": False}
|
|
body, more = chunks.pop(0)
|
|
return {"type": "http.request", "body": body, "more_body": more}
|
|
|
|
messages = []
|
|
|
|
async def send(message):
|
|
messages.append(message)
|
|
|
|
async def run():
|
|
await BodyCapMiddleware(_dummy_app)(scope, receive, send)
|
|
|
|
asyncio.run(run())
|
|
return next(m["status"] for m in messages if m["type"] == "http.response.start")
|
|
|
|
|
|
def test_content_length_malformat_nu_crapa(monkeypatch):
|
|
monkeypatch.setenv("AUTOPASS_MAX_REQUEST_BYTES", "1000")
|
|
from app.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
try:
|
|
status = _run_body_cap(
|
|
headers=[(b"content-length", b"abc")],
|
|
body_chunks=[(b"body mic", False)],
|
|
)
|
|
assert status == 200
|
|
finally:
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_content_length_negativ_nu_crapa_dar_plafonul_ramane_aplicat(monkeypatch):
|
|
monkeypatch.setenv("AUTOPASS_MAX_REQUEST_BYTES", "1000")
|
|
from app.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
try:
|
|
status = _run_body_cap(
|
|
headers=[(b"content-length", b"-5")],
|
|
body_chunks=[(b"x" * 2000, False)],
|
|
)
|
|
assert status == 413
|
|
finally:
|
|
get_settings.cache_clear()
|