342 lines
12 KiB
Python
342 lines
12 KiB
Python
"""Regression tests for the hardening pass (security, robustness, data-quality edge cases)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import UTC, datetime
|
|
from threading import Barrier
|
|
|
|
import pytest
|
|
from sqlalchemy import delete, func, select
|
|
|
|
from app.core.config import Settings, get_settings, insecure_default_secrets
|
|
from app.core.db import SessionLocal
|
|
from app.core.errors import AppError
|
|
from app.core.observability import UNMATCHED_ROUTE_LABEL
|
|
from app.core.ratelimit import FailedAttemptLimiter, SlidingWindowLimiter
|
|
from app.models.audit import AuditEvent
|
|
from app.models.customer import Customer
|
|
from app.models.data_quality import DataQualityIssue
|
|
from app.schemas import CurrentUser
|
|
from app.services.data_quality import defer_issue, run_scan
|
|
from tests.test_return import _activate_booking, _return_body
|
|
|
|
|
|
def test_insecure_defaults_include_mounted_mcp_boundary_even_when_status_check_is_disabled():
|
|
defaults = Settings(_env_file=None)
|
|
assert "app_secret" in insecure_default_secrets(defaults)
|
|
assert "mcp_hub_service_token" in insecure_default_secrets(defaults)
|
|
hardened = Settings(
|
|
_env_file=None,
|
|
app_secret="x" * 32,
|
|
n8n_callback_token="c" * 32,
|
|
mcp_hub_service_token="m" * 32,
|
|
)
|
|
assert insecure_default_secrets(hardened) == []
|
|
|
|
|
|
def test_production_refuses_to_start_with_placeholder_secrets(monkeypatch):
|
|
monkeypatch.setenv("MOBILITYOPS_ENV", "production")
|
|
monkeypatch.setenv("APP_SECRET", "replace-in-production")
|
|
get_settings.cache_clear()
|
|
try:
|
|
with pytest.raises(RuntimeError, match="placeholder secrets"):
|
|
get_settings()
|
|
finally:
|
|
get_settings.cache_clear()
|
|
monkeypatch.undo()
|
|
get_settings.cache_clear()
|
|
# The cached settings the running app relies on must be intact afterwards.
|
|
assert get_settings().mobilityops_env == "test"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("public_url", "secure_cookie", "message"),
|
|
[
|
|
("http://fleetops.example.test", "true", "must use HTTPS"),
|
|
("https://fleetops.example.test", "false", "must be true"),
|
|
],
|
|
)
|
|
def test_production_refuses_cleartext_or_insecure_session_cookie(
|
|
monkeypatch, public_url, secure_cookie, message
|
|
):
|
|
values = {
|
|
"MOBILITYOPS_ENV": "production",
|
|
"APP_SECRET": "a" * 32,
|
|
"N8N_CALLBACK_TOKEN": "c" * 32,
|
|
"MCP_HUB_SERVICE_TOKEN": "m" * 32,
|
|
"MOBILITYOPS_PUBLIC_URL": public_url,
|
|
"SESSION_COOKIE_SECURE": secure_cookie,
|
|
}
|
|
for name, value in values.items():
|
|
monkeypatch.setenv(name, value)
|
|
get_settings.cache_clear()
|
|
try:
|
|
with pytest.raises(RuntimeError, match=message):
|
|
get_settings()
|
|
finally:
|
|
get_settings.cache_clear()
|
|
monkeypatch.undo()
|
|
get_settings.cache_clear()
|
|
assert get_settings().mobilityops_env == "test"
|
|
|
|
|
|
def test_failed_attempt_limiter_blocks_after_threshold_and_resets():
|
|
limiter = FailedAttemptLimiter(max_failures=3, window_seconds=60)
|
|
for _ in range(3):
|
|
assert limiter.retry_after_seconds("1.2.3.4") == 0
|
|
limiter.record_failure("1.2.3.4")
|
|
assert limiter.retry_after_seconds("1.2.3.4") > 0
|
|
assert limiter.retry_after_seconds("5.6.7.8") == 0
|
|
limiter.reset("1.2.3.4")
|
|
assert limiter.retry_after_seconds("1.2.3.4") == 0
|
|
|
|
|
|
def test_sliding_window_limiter_counts_successes_and_isolates_keys():
|
|
limiter = SlidingWindowLimiter(max_requests=2, window_seconds=60)
|
|
assert limiter.consume("session-a") == 0
|
|
assert limiter.consume("session-a") == 0
|
|
assert limiter.consume("session-a") > 0
|
|
assert limiter.consume("session-b") == 0
|
|
|
|
|
|
def test_login_client_key_uses_proxy_appended_peer_not_spoofed_leading_value():
|
|
from starlette.requests import Request
|
|
|
|
from app.api.routers.auth import _client_key
|
|
|
|
request = Request(
|
|
{
|
|
"type": "http",
|
|
"headers": [(b"x-forwarded-for", b"203.0.113.99, 198.51.100.7")],
|
|
"client": ("172.20.0.3", 12345),
|
|
}
|
|
)
|
|
assert _client_key(request) == "198.51.100.7"
|
|
|
|
|
|
def test_audit_export_accepts_naive_datetimes(ops_client):
|
|
response = ops_client.get(
|
|
"/api/v1/audit/export.csv",
|
|
params={"occurred_from": "2026-01-01T00:00:00", "occurred_to": "2026-01-15T00:00:00"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_audit_list_rejects_non_uuid_correlation_id(ops_client):
|
|
response = ops_client.get("/api/v1/audit", params={"correlation_id": "not-a-uuid"})
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_unmatched_routes_do_not_create_metric_series(client):
|
|
probe = f"/api/v1/does-not-exist-{uuid.uuid4().hex}"
|
|
assert client.get(probe).status_code == 404
|
|
metrics = client.get("/metrics").text
|
|
assert probe not in metrics
|
|
assert UNMATCHED_ROUTE_LABEL in metrics
|
|
|
|
|
|
def test_bookings_page_for_unknown_vehicle_keeps_page_shape(ops_client):
|
|
response = ops_client.get("/api/v1/bookings", params={"vehicle_ref": "MO-NOPE", "page": 1})
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"items": [],
|
|
"page": 1,
|
|
"page_size": 25,
|
|
"total": 0,
|
|
"total_pages": 1,
|
|
}
|
|
|
|
|
|
def test_return_idempotency_key_rejects_different_body(ops_client):
|
|
booking_ref = _activate_booking("MO-011", start_odometer_km=30000)
|
|
key = "test-return-fingerprint-001"
|
|
first = ops_client.post(
|
|
f"/api/v1/bookings/{booking_ref}/return",
|
|
json=_return_body(end_odometer_km=30500),
|
|
headers={"Idempotency-Key": key},
|
|
)
|
|
assert first.status_code == 201
|
|
replay = ops_client.post(
|
|
f"/api/v1/bookings/{booking_ref}/return",
|
|
json=_return_body(end_odometer_km=30500),
|
|
headers={"Idempotency-Key": key},
|
|
)
|
|
assert replay.status_code == 201
|
|
mismatch = ops_client.post(
|
|
f"/api/v1/bookings/{booking_ref}/return",
|
|
json=_return_body(end_odometer_km=30999),
|
|
headers={"Idempotency-Key": key},
|
|
)
|
|
assert mismatch.status_code == 409
|
|
assert mismatch.json()["error"]["code"] == "IDEMPOTENCY_KEY_REUSED"
|
|
|
|
|
|
def test_return_callback_rejects_malformed_correlation_id(client):
|
|
event_id = str(uuid.uuid4())
|
|
response = client.post(
|
|
"/api/v1/integrations/n8n/return-callback",
|
|
json={"event_id": event_id, "follow_up": "cleaning", "correlation_id": "nope"},
|
|
headers={
|
|
"Idempotency-Key": event_id,
|
|
"X-Service-Token": get_settings().n8n_callback_token,
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_blocked_checkout_booking_can_be_cancelled(ops_client):
|
|
window = {"starts_at": "2051-03-01T10:00:00Z", "ends_at": "2051-03-02T12:00:00Z"}
|
|
available = ops_client.get("/api/v1/bookings/availability", params=window).json()
|
|
vehicle_option = next(item for item in available if item["operational_status"] == "available")
|
|
vehicle = ops_client.get(f"/api/v1/vehicles/{vehicle_option['public_ref']}").json()
|
|
booking = ops_client.post(
|
|
"/api/v1/bookings",
|
|
json={
|
|
"customer_ref": "CUS-0002",
|
|
"vehicle_ref": vehicle["public_ref"],
|
|
**window,
|
|
},
|
|
).json()
|
|
confirmed = ops_client.post(
|
|
f"/api/v1/bookings/{booking['public_ref']}/complete-requirements",
|
|
json={"confirmation": "Licence and rental conditions checked"},
|
|
)
|
|
assert confirmed.status_code == 200
|
|
checkout = ops_client.post(
|
|
f"/api/v1/bookings/{booking['public_ref']}/checkout",
|
|
json={
|
|
"start_odometer_km": vehicle["odometer_km"],
|
|
"fuel_level_percent": 80,
|
|
"cleanliness_ok": True,
|
|
"damage_reported": True,
|
|
"technical_warning": False,
|
|
},
|
|
)
|
|
assert checkout.status_code == 200
|
|
assert checkout.json()["booking_status"] == "blocked"
|
|
cancelled = ops_client.post(
|
|
f"/api/v1/bookings/{booking['public_ref']}/cancel",
|
|
json={"reason": "Vehicle damaged at departure inspection"},
|
|
)
|
|
assert cancelled.status_code == 200
|
|
assert cancelled.json()["status"] == "cancelled"
|
|
|
|
|
|
def test_scan_does_not_flag_anonymised_customers_as_missing_fields(ops_client):
|
|
with SessionLocal() as db:
|
|
customer = Customer(
|
|
public_ref="CUS-ANON-SCAN",
|
|
first_name="Anoniem",
|
|
last_name="Klant",
|
|
email=None,
|
|
phone=None,
|
|
postal_code=None,
|
|
city=None,
|
|
date_of_birth=None,
|
|
anonymized_at=datetime.now(UTC),
|
|
)
|
|
db.add(customer)
|
|
db.commit()
|
|
customer_id = customer.id
|
|
try:
|
|
with SessionLocal() as db:
|
|
run_scan(db)
|
|
db.commit()
|
|
flagged = db.scalar(
|
|
select(DataQualityIssue.id).where(
|
|
DataQualityIssue.entity_type == "customer",
|
|
DataQualityIssue.entity_id == customer_id,
|
|
)
|
|
)
|
|
assert flagged is None
|
|
finally:
|
|
with SessionLocal() as db:
|
|
db.execute(delete(DataQualityIssue).where(DataQualityIssue.entity_id == customer_id))
|
|
db.execute(delete(AuditEvent).where(AuditEvent.entity_id == customer_id))
|
|
db.execute(delete(Customer).where(Customer.id == customer_id))
|
|
db.commit()
|
|
|
|
|
|
def test_concurrent_quality_scans_leave_only_one_open_issue_per_condition():
|
|
barrier = Barrier(2)
|
|
|
|
def scan() -> None:
|
|
with SessionLocal() as db:
|
|
barrier.wait()
|
|
run_scan(db)
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
list(executor.map(lambda _index: scan(), range(2)))
|
|
|
|
with SessionLocal() as db:
|
|
duplicates = db.execute(
|
|
select(
|
|
DataQualityIssue.rule_type,
|
|
DataQualityIssue.entity_type,
|
|
DataQualityIssue.entity_id,
|
|
func.count(DataQualityIssue.id),
|
|
)
|
|
.where(DataQualityIssue.status == "open")
|
|
.group_by(
|
|
DataQualityIssue.rule_type,
|
|
DataQualityIssue.entity_type,
|
|
DataQualityIssue.entity_id,
|
|
)
|
|
.having(func.count(DataQualityIssue.id) > 1)
|
|
).all()
|
|
assert duplicates == []
|
|
|
|
|
|
def test_concurrent_issue_resolution_records_exactly_one_decision():
|
|
issue_id = uuid.uuid4()
|
|
issue_ref = f"DQ-RACE-{uuid.uuid4().hex[:8].upper()}"
|
|
with SessionLocal() as db:
|
|
db.add(
|
|
DataQualityIssue(
|
|
id=issue_id,
|
|
public_ref=issue_ref,
|
|
rule_type="missing_required_field",
|
|
entity_type="customer",
|
|
entity_id=uuid.uuid4(),
|
|
severity="low",
|
|
status="open",
|
|
evidence_json={},
|
|
proposed_action_json={},
|
|
detected_at=datetime.now(UTC),
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
actor = CurrentUser(
|
|
public_ref="USR-RACE", display_name="Race Manager", role="operations_manager"
|
|
)
|
|
barrier = Barrier(2)
|
|
|
|
def resolve() -> str:
|
|
with SessionLocal() as db:
|
|
barrier.wait()
|
|
try:
|
|
return defer_issue(db, issue_ref, actor).status
|
|
except AppError as exc:
|
|
return exc.code
|
|
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
outcomes = list(executor.map(lambda _index: resolve(), range(2)))
|
|
assert sorted(outcomes) == ["ISSUE_NOT_OPEN", "deferred"]
|
|
with SessionLocal() as db:
|
|
decisions = db.scalar(
|
|
select(func.count(AuditEvent.id)).where(
|
|
AuditEvent.entity_id == issue_id,
|
|
AuditEvent.action == "data_quality_issue_deferred",
|
|
)
|
|
)
|
|
assert decisions == 1
|
|
finally:
|
|
with SessionLocal() as db:
|
|
db.execute(delete(AuditEvent).where(AuditEvent.entity_id == issue_id))
|
|
db.execute(delete(DataQualityIssue).where(DataQualityIssue.id == issue_id))
|
|
db.commit()
|