Files
Jens faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
Initial public release
2026-08-31 21:56:53 +02:00

507 lines
20 KiB
Python

from __future__ import annotations
from pathlib import Path
from types import SimpleNamespace
from uuid import UUID, uuid4
from fastapi.testclient import TestClient
from app.core.config import get_settings
from app.core.public_demo import PUBLIC_DEMO_PROJECT_ID
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.change_detection_service import ChangeDetectionService
from app.services.dataset_service import DatasetService
from app.services.detection_service import DetectionService
from app.services.demo_workflow_service import DemoWorkflowService
from app.services.raster_operations_service import RasterOperationsService
from app.services.segmentation_service import SegmentationService
from app.services.job_service import JobService
def auth_client(
monkeypatch,
*,
guest_access: bool = False,
require_https: bool = False,
base_url: str = "http://testserver",
) -> 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_REQUIRE_HTTPS", "true" if require_https else "false")
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(), base_url=base_url)
def test_guest_access_defaults_off_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 False
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,
"authentik_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,
"authentik_enabled": False,
"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_operator_login_can_require_https(monkeypatch) -> None:
insecure_client = auth_client(monkeypatch, require_https=True)
rejected = insecure_client.post(
"/api/v1/auth/login",
json={"username": "operator", "password": "correct horse battery staple"},
)
spoofed = insecure_client.post(
"/api/v1/auth/login",
headers={"x-forwarded-proto": "https", "x-real-ip": "203.0.113.9"},
json={"username": "operator", "password": "correct horse battery staple"},
)
secure_client = auth_client(
monkeypatch,
require_https=True,
base_url="https://testserver",
)
accepted = secure_client.post(
"/api/v1/auth/login",
json={"username": "operator", "password": "correct horse battery staple"},
)
assert rejected.status_code == 426
assert rejected.json()["error"] == "AUTH_HTTPS_REQUIRED"
assert spoofed.status_code == 426
assert accepted.status_code == 200
assert "secure" in accepted.headers["set-cookie"].lower()
def test_guest_login_exposes_models_but_rejects_management_and_cross_project_requests(monkeypatch) -> None:
project_id = PUBLIC_DEMO_PROJECT_ID
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")
global_source_registry = client.get("/api/v1/source-registry/grb")
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": [],
},
)
bounded_acquisition = client.post(
f"/api/v1/projects/{project_id}/datasets/orthophoto/acquire",
json={},
)
cross_project_acquisition = client.post(
"/api/v1/projects/00000000-0000-0000-0000-000000000999/datasets/orthophoto/acquire",
json={},
)
bounded_derived_selection = client.post(
f"/api/v1/projects/{project_id}/datasets/{demo.candidate_dataset_id}/vector/select/derive",
json={},
)
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 global_source_registry.status_code == 403
assert global_source_registry.json()["error"] == "GUEST_ROUTE_NOT_AVAILABLE"
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"
assert bounded_acquisition.status_code == 422
assert bounded_acquisition.json()["error"] != "GUEST_READ_ONLY"
assert cross_project_acquisition.status_code == 403
assert cross_project_acquisition.json()["error"] == "GUEST_PROJECT_SCOPE_REQUIRED"
assert bounded_derived_selection.status_code == 422
assert bounded_derived_selection.json()["error"] != "GUEST_READ_ONLY"
def test_guest_change_detection_binds_both_datasets_to_signed_demo_project(monkeypatch) -> None:
project_id = PUBLIC_DEMO_PROJECT_ID
other_project_id = UUID("00000000-0000-0000-0000-000000000999")
source_dataset_id = UUID("00000000-0000-0000-0000-000000000125")
target_dataset_id = UUID("00000000-0000-0000-0000-000000000126")
cross_project_dataset_id = UUID("00000000-0000-0000-0000-000000000998")
demo = DemoWorkflowResponse(
project_id=project_id,
area_id=UUID("00000000-0000-0000-0000-000000000124"),
reference_dataset_id=source_dataset_id,
candidate_dataset_id=target_dataset_id,
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))
class FakeDb:
def get(self, _model, dataset_id):
bound_project_id = other_project_id if dataset_id == cross_project_dataset_id else project_id
return SimpleNamespace(id=dataset_id, project_id=bound_project_id, dataset_type="vector")
validated_datasets: list[tuple[UUID, UUID, str]] = []
def validate_dataset(_db, dataset_id, requested_project_id, label):
validated_datasets.append((dataset_id, requested_project_id, label))
return SimpleNamespace(id=dataset_id, project_id=requested_project_id, dataset_type="vector")
monkeypatch.setattr(
ChangeDetectionService,
"_get_project_vector_dataset",
staticmethod(validate_dataset),
)
monkeypatch.setattr(
JobService,
"run_sync_job",
staticmethod(
lambda **kwargs: SimpleNamespace(
id=uuid4(),
job_type=kwargs["job_type"],
status="success",
project_id=kwargs["project_id"],
dataset_id=source_dataset_id,
input_dataset_id=source_dataset_id,
output_dataset_id=None,
parameters_json=kwargs["parameters"],
result_json={},
error_message=None,
created_at=None,
started_at=None,
finished_at=None,
)
),
)
client = auth_client(monkeypatch, guest_access=True)
def fake_db():
yield FakeDb()
client.app.dependency_overrides[get_db] = fake_db
assert client.post("/api/v1/auth/guest").status_code == 200
accepted = client.post(
"/api/v1/analysis/change-detection",
json={
"source_dataset_id": str(source_dataset_id),
"target_dataset_id": str(target_dataset_id),
},
)
rejected = client.post(
"/api/v1/analysis/change-detection",
json={
"source_dataset_id": str(cross_project_dataset_id),
"target_dataset_id": str(target_dataset_id),
},
)
assert accepted.status_code == 200
assert accepted.json()["data"]["project_id"] == str(project_id)
assert validated_datasets == [
(source_dataset_id, project_id, "Source"),
(target_dataset_id, project_id, "Target"),
]
assert rejected.status_code == 403
assert rejected.json()["error"] == "GUEST_PROJECT_SCOPE_REQUIRED"
def test_guest_can_prepare_tiles_and_queue_project_scoped_detection(monkeypatch) -> None:
project_id = PUBLIC_DEMO_PROJECT_ID
raster_dataset_id = UUID("00000000-0000-0000-0000-000000000127")
manifest_path = "/app/storage/tiles/demo/manifest.json"
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=raster_dataset_id,
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))
monkeypatch.setattr(
DatasetService,
"get_dataset",
staticmethod(lambda _db, _dataset_id: SimpleNamespace(project_id=project_id)),
)
def job(*, job_type: str, result_json: dict | None = None):
return SimpleNamespace(
id=uuid4(),
job_type=job_type,
status="success" if result_json else "queued",
project_id=project_id,
dataset_id=raster_dataset_id,
input_dataset_id=raster_dataset_id,
output_dataset_id=None,
parameters_json={},
result_json=result_json,
error_message=None,
created_at=None,
started_at=None,
finished_at=None,
)
tile_parameters: dict = {}
def tile(_db, _dataset_id, **kwargs):
tile_parameters.update(kwargs)
return {"manifest_path": manifest_path}
monkeypatch.setattr(RasterOperationsService, "tile", staticmethod(tile))
monkeypatch.setattr(
"app.api.routes.datasets._run_job_sync",
lambda **kwargs: job(job_type="raster.tile", result_json=kwargs["operation"]()),
)
queued_parameters: dict = {}
def enqueue_detection(**kwargs):
queued_parameters.update(kwargs)
return job(job_type="detection.run")
monkeypatch.setattr(DetectionService, "enqueue_detection", staticmethod(enqueue_detection))
queued_segmentation_parameters: dict = {}
def enqueue_segmentation(**kwargs):
queued_segmentation_parameters.update(kwargs)
return job(job_type="segmentation.run")
monkeypatch.setattr(SegmentationService, "enqueue_segmentation", staticmethod(enqueue_segmentation))
client = auth_client(monkeypatch, guest_access=True)
def fake_db():
yield object()
client.app.dependency_overrides[get_db] = fake_db
assert client.post("/api/v1/auth/guest").status_code == 200
tile_response = client.post(
f"/api/v1/projects/{project_id}/datasets/{raster_dataset_id}/raster/tile",
json={"tile_size": 512, "overlap": 64},
)
detection_response = client.post(
f"/api/v1/detection/run-async?project_id={project_id}",
json={
"project_id": str(project_id),
"dataset_id": str(raster_dataset_id),
"model_id": "yolo-configured",
"model_asset_id": "active-model",
"confidence_threshold": 0.15,
"tile_manifest_path": manifest_path,
"parameters_json": {},
},
)
segmentation_response = client.post(
f"/api/v1/segmentation/run-async?project_id={project_id}",
json={
"project_id": str(project_id),
"dataset_id": str(raster_dataset_id),
"model_id": "sam-configured",
"confidence_threshold": 0.5,
"tile_manifest_path": manifest_path,
"parameters_json": {},
},
)
assert tile_response.status_code == 201
assert tile_response.json()["data"]["result_json"]["manifest_path"] == manifest_path
assert detection_response.status_code == 200
assert detection_response.json()["data"]["status"] == "queued"
assert segmentation_response.status_code == 200
assert segmentation_response.json()["data"]["status"] == "queued"
assert queued_parameters["project_id"] == project_id
assert queued_parameters["dataset_id"] == raster_dataset_id
assert queued_parameters["tile_manifest_path"] == manifest_path
assert queued_segmentation_parameters["project_id"] == project_id
assert queued_segmentation_parameters["dataset_id"] == raster_dataset_id
assert queued_segmentation_parameters["tile_manifest_path"] == manifest_path
assert tile_parameters["max_tiles"] == get_settings().yolo_max_tiles
def test_guest_session_tokens_fail_closed_without_a_project_scope(monkeypatch) -> None:
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 '-e GEOINTEL_AUTHENTIK_CLIENT_SECRET="$GEOINTEL_AUTHENTIK_CLIENT_SECRET"' in runner
assert "GEOINTEL_AUTH_PASSWORD_HASH=" in example
assert "GEOINTEL_AUTHENTIK_CLIENT_SECRET=" in example
assert "GEOINTEL_AUTH_PASSWORD=" not in runner
assert "GEOINTEL_AUTH_ENABLED=true" in example
assert "GEOINTEL_AUTH_REQUIRE_HTTPS=true" in example
assert "GEOINTEL_GUEST_ACCESS_ENABLED=false" in example
assert 'GEOINTEL_AUTH_ENABLED="${GEOINTEL_AUTH_ENABLED:-true}"' in runner
assert 'GEOINTEL_AUTH_REQUIRE_HTTPS="${GEOINTEL_AUTH_REQUIRE_HTTPS:-true}"' in runner
assert 'GEOINTEL_GUEST_ACCESS_ENABLED="${GEOINTEL_GUEST_ACCESS_ENABLED:-false}"' in runner
assert "/api/v1/auth/session" in browser_smoke