228 lines
9.4 KiB
Python
228 lines
9.4 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from uuid import UUID
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.config import get_settings
|
|
from app.db.session import get_db
|
|
from app.main import create_app
|
|
from app.schemas.demo import DemoWorkflowResponse
|
|
from app.services.auth_service import AuthService
|
|
from app.services.demo_workflow_service import DemoWorkflowService
|
|
|
|
|
|
def auth_client(monkeypatch, *, guest_access: bool = False) -> TestClient:
|
|
password_hash = AuthService.hash_password(
|
|
"correct horse battery staple",
|
|
salt=b"geointel-test-salt",
|
|
iterations=100_000,
|
|
)
|
|
monkeypatch.setenv("GEOINTEL_AUTH_ENABLED", "true")
|
|
monkeypatch.setenv("GEOINTEL_AUTH_USERNAME", "operator")
|
|
monkeypatch.setenv("GEOINTEL_AUTH_PASSWORD_HASH", password_hash)
|
|
monkeypatch.setenv("GEOINTEL_AUTH_SESSION_SECRET", "test-session-secret-that-is-long-enough")
|
|
monkeypatch.setenv("GEOINTEL_GUEST_ACCESS_ENABLED", "true" if guest_access else "false")
|
|
monkeypatch.setenv("GEOINTEL_GUEST_DISPLAY_NAME", "Gast")
|
|
monkeypatch.setenv("GEOINTEL_GUEST_SESSION_TTL_SECONDS", "7200")
|
|
return TestClient(create_app())
|
|
|
|
|
|
def test_guest_access_defaults_on_when_operator_authentication_is_enabled(monkeypatch) -> None:
|
|
password_hash = AuthService.hash_password(
|
|
"correct horse battery staple",
|
|
salt=b"geointel-test-salt",
|
|
iterations=100_000,
|
|
)
|
|
monkeypatch.setenv("GEOINTEL_AUTH_ENABLED", "true")
|
|
monkeypatch.setenv("GEOINTEL_AUTH_USERNAME", "operator")
|
|
monkeypatch.setenv("GEOINTEL_AUTH_PASSWORD_HASH", password_hash)
|
|
monkeypatch.setenv("GEOINTEL_AUTH_SESSION_SECRET", "test-session-secret-that-is-long-enough")
|
|
monkeypatch.delenv("GEOINTEL_GUEST_ACCESS_ENABLED", raising=False)
|
|
|
|
client = TestClient(create_app())
|
|
session = client.get("/api/v1/auth/session")
|
|
|
|
assert session.status_code == 200
|
|
assert session.json()["data"]["authentication_required"] is True
|
|
assert session.json()["data"]["guest_access_enabled"] is True
|
|
|
|
|
|
def test_guest_default_is_inactive_but_valid_when_operator_authentication_is_disabled(monkeypatch) -> None:
|
|
monkeypatch.setenv("GEOINTEL_AUTH_ENABLED", "false")
|
|
monkeypatch.delenv("GEOINTEL_AUTH_USERNAME", raising=False)
|
|
monkeypatch.delenv("GEOINTEL_AUTH_PASSWORD_HASH", raising=False)
|
|
monkeypatch.delenv("GEOINTEL_AUTH_SESSION_SECRET", raising=False)
|
|
monkeypatch.delenv("GEOINTEL_GUEST_ACCESS_ENABLED", raising=False)
|
|
|
|
client = TestClient(create_app())
|
|
session = client.get("/api/v1/auth/session")
|
|
|
|
assert session.status_code == 200
|
|
assert session.json()["data"]["authentication_required"] is False
|
|
assert session.json()["data"]["authenticated"] is True
|
|
assert session.json()["data"]["guest_access_enabled"] is False
|
|
|
|
|
|
def test_auth_session_and_health_are_public_but_api_is_protected(monkeypatch) -> None:
|
|
client = auth_client(monkeypatch)
|
|
|
|
session = client.get("/api/v1/auth/session")
|
|
protected = client.get("/api/v1/protected-probe")
|
|
health = client.get("/health/live")
|
|
|
|
assert session.status_code == 200
|
|
assert session.json()["data"] == {
|
|
"authentication_required": True,
|
|
"authenticated": False,
|
|
"username": None,
|
|
"expires_at": None,
|
|
"role": None,
|
|
"guest_access_enabled": False,
|
|
"guest_project_id": None,
|
|
}
|
|
assert protected.status_code == 401
|
|
assert protected.json()["error"] == "AUTHENTICATION_REQUIRED"
|
|
assert health.status_code == 200
|
|
|
|
|
|
def test_login_uses_http_only_session_cookie_and_logout_revokes_browser_access(monkeypatch) -> None:
|
|
client = auth_client(monkeypatch, guest_access=True)
|
|
|
|
invalid = client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "operator", "password": "wrong"},
|
|
)
|
|
login = client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "operator", "password": "correct horse battery staple"},
|
|
)
|
|
authenticated = client.get("/api/v1/auth/session")
|
|
protected_after_login = client.get("/api/v1/protected-probe")
|
|
logout = client.post("/api/v1/auth/logout")
|
|
protected_after_logout = client.get("/api/v1/protected-probe")
|
|
|
|
assert invalid.status_code == 401
|
|
assert invalid.json()["error"] == "INVALID_CREDENTIALS"
|
|
assert login.status_code == 200
|
|
assert login.json()["data"] == {
|
|
"authentication_required": True,
|
|
"authenticated": True,
|
|
"username": "operator",
|
|
"expires_at": login.json()["data"]["expires_at"],
|
|
"role": "operator",
|
|
"guest_access_enabled": True,
|
|
"guest_project_id": None,
|
|
}
|
|
cookie = login.headers["set-cookie"].lower()
|
|
assert "httponly" in cookie
|
|
assert "samesite=strict" in cookie
|
|
assert authenticated.json()["data"]["authenticated"] is True
|
|
assert authenticated.json()["data"]["role"] == "operator"
|
|
assert protected_after_login.status_code == 404
|
|
assert logout.status_code == 200
|
|
assert logout.json()["data"]["guest_access_enabled"] is True
|
|
assert protected_after_logout.status_code == 401
|
|
|
|
|
|
def test_guest_login_exposes_models_but_rejects_management_and_cross_project_requests(monkeypatch) -> None:
|
|
project_id = UUID("00000000-0000-0000-0000-000000000123")
|
|
demo = DemoWorkflowResponse(
|
|
project_id=project_id,
|
|
area_id=UUID("00000000-0000-0000-0000-000000000124"),
|
|
reference_dataset_id=UUID("00000000-0000-0000-0000-000000000125"),
|
|
candidate_dataset_id=UUID("00000000-0000-0000-0000-000000000126"),
|
|
raster_dataset_id=UUID("00000000-0000-0000-0000-000000000127"),
|
|
quality_check_id=UUID("00000000-0000-0000-0000-000000000128"),
|
|
metric_count=6,
|
|
status="ok",
|
|
message="Demo ready",
|
|
created=False,
|
|
)
|
|
monkeypatch.setattr(DemoWorkflowService, "seed", staticmethod(lambda _db: demo))
|
|
client = auth_client(monkeypatch, guest_access=True)
|
|
|
|
def fake_db():
|
|
yield object()
|
|
|
|
client.app.dependency_overrides[get_db] = fake_db
|
|
|
|
guest_login = client.post("/api/v1/auth/guest")
|
|
guest_session = client.get("/api/v1/auth/session")
|
|
mutation = client.post("/api/v1/projects", json={"name": "Not allowed"})
|
|
other_project = client.get("/api/v1/projects/00000000-0000-0000-0000-000000000999")
|
|
detection_models = client.get("/api/v1/detection/models")
|
|
segmentation_models = client.get("/api/v1/segmentation/models")
|
|
cross_project_runs = client.get(
|
|
"/api/v1/detection/runs?project_id=00000000-0000-0000-0000-000000000999"
|
|
)
|
|
cross_project_coverage = client.post(
|
|
"/api/v1/external/coverage/resolve",
|
|
json={
|
|
"project_id": "00000000-0000-0000-0000-000000000999",
|
|
"bbox": {"minx": 4.9, "miny": 51.0, "maxx": 5.0, "maxy": 51.1},
|
|
"themes": [],
|
|
},
|
|
)
|
|
|
|
assert guest_login.status_code == 200
|
|
assert guest_login.json()["data"]["role"] == "guest"
|
|
assert guest_login.json()["data"]["username"] == "Gast"
|
|
assert guest_login.json()["data"]["guest_project_id"] == str(project_id)
|
|
assert "httponly" in guest_login.headers["set-cookie"].lower()
|
|
assert guest_session.json()["data"]["role"] == "guest"
|
|
assert mutation.status_code == 403
|
|
assert mutation.json()["error"] == "GUEST_READ_ONLY"
|
|
assert other_project.status_code == 403
|
|
assert other_project.json()["error"] == "GUEST_PROJECT_SCOPE_REQUIRED"
|
|
assert detection_models.status_code == 200
|
|
assert detection_models.json()["data"]["models"]
|
|
assert segmentation_models.status_code == 200
|
|
assert cross_project_runs.status_code == 403
|
|
assert cross_project_runs.json()["error"] == "GUEST_PROJECT_SCOPE_REQUIRED"
|
|
assert cross_project_coverage.status_code == 403
|
|
assert cross_project_coverage.json()["error"] == "GUEST_PROJECT_SCOPE_REQUIRED"
|
|
|
|
|
|
def test_guest_session_tokens_fail_closed_without_a_project_scope(monkeypatch) -> None:
|
|
client = auth_client(monkeypatch, guest_access=True)
|
|
settings = get_settings()
|
|
|
|
try:
|
|
AuthService.create_session_token("Gast", settings, role="guest")
|
|
except ValueError as error:
|
|
assert "demo project" in str(error)
|
|
else: # pragma: no cover - defensive assertion
|
|
raise AssertionError("An unscoped guest token should not be created")
|
|
|
|
|
|
def test_password_hash_and_session_signatures_fail_closed(monkeypatch) -> None:
|
|
client = auth_client(monkeypatch)
|
|
login = client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "operator", "password": "correct horse battery staple"},
|
|
)
|
|
token = login.cookies.get("geointel_session")
|
|
|
|
assert token
|
|
client.cookies.set("geointel_session", f"{token}tampered")
|
|
session = client.get("/api/v1/auth/session")
|
|
|
|
assert session.status_code == 200
|
|
assert session.json()["data"]["authenticated"] is False
|
|
|
|
|
|
def test_unraid_runtime_carries_only_hashed_operator_credentials() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
runner = (root / "deploy/unraid/run-dockerman-container.sh").read_text(encoding="utf-8")
|
|
example = (root / "deploy/unraid/geointel.env.example").read_text(encoding="utf-8")
|
|
browser_smoke = (root / "scripts/verify_browser_runtime.sh").read_text(encoding="utf-8")
|
|
|
|
assert '-e GEOINTEL_AUTH_PASSWORD_HASH="$GEOINTEL_AUTH_PASSWORD_HASH"' in runner
|
|
assert "GEOINTEL_AUTH_PASSWORD_HASH=" in example
|
|
assert "GEOINTEL_AUTH_PASSWORD=" not in runner
|
|
assert "GEOINTEL_GUEST_ACCESS_ENABLED=true" in example
|
|
assert 'GEOINTEL_GUEST_ACCESS_ENABLED="${GEOINTEL_GUEST_ACCESS_ENABLED:-true}"' in runner
|
|
assert "/api/v1/auth/session" in browser_smoke
|