kyleandClaude Opus 5.5
2fa9eea274
nginx now checks every dashboard and API request with oauth2-proxy (auth_request), which signs users in with Microsoft Entra ID. Who may sign in is controlled in Entra: a single-tenant app registration with "Assignment required", so only assigned users/groups get through. - deploy/oauth2-proxy.cfg (entra-id provider, auth-only mode, secure cookie, trusted proxy = localhost) and deploy/oauth2-proxy.service - nginx: pages redirect to Microsoft sign-in and back to the same URL; API calls get a plain 401 (the page already asks to reload); /api/v1/ingest stays key-only for collectors; /oauth2/auth is internal; the signed-in user is passed to the API in headers nginx sets (so browsers can't spoof them); dashboard files are served with Cache-Control: no-cache so updates show up at once - GET /api/v1/me; the top bar shows the signed-in user and a Sign out that also signs out of Microsoft (otherwise SSO signs straight back in) - README: Entra and server setup steps Tested locally with oauth2-proxy 7.15.5 and nginx 1.31 running the deploy configs (TLS, self-signed): redirects to the tenant's sign-in with the return URL kept, 401s, ingest bypass, sign-out allow-list, cookie flags, security headers. The signed-in path was tested with a stand-in for oauth2-proxy's auth answer; a real Entra sign-in wasn't. Co-Authored-By: Claude Opus 5.5 <[email protected]>
131 lines
5.7 KiB
Python
131 lines
5.7 KiB
Python
from datetime import timedelta
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import db
|
|
from app.auth import generate_key, hash_key
|
|
from app.main import create_app
|
|
from app.settings import Settings
|
|
from app.timeutil import to_db, utcnow
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path):
|
|
settings = Settings(db_path=tmp_path / "test.db")
|
|
with TestClient(create_app(settings)) as c:
|
|
key = generate_key()
|
|
conn = db.connect(settings.db_path)
|
|
conn.execute("INSERT INTO api_keys (name, key_hash, created_at) VALUES ('test', ?, ?)",
|
|
(hash_key(key), to_db(utcnow())))
|
|
conn.close()
|
|
c.headers["Authorization"] = f"Bearer {key}"
|
|
yield c
|
|
|
|
|
|
def iso(delta: timedelta) -> str:
|
|
return (utcnow() + delta).isoformat()
|
|
|
|
|
|
def payload(run=None, **job):
|
|
body = {"tenant": "Acme Corp", "source_system": "Veeam B&R",
|
|
"job": {"external_id": "job-1", "name": "File Server", **job}}
|
|
if run is not None:
|
|
body["run"] = run
|
|
return body
|
|
|
|
|
|
def test_ingest_requires_valid_key(client):
|
|
assert client.post("/api/v1/ingest", json=payload(), headers={"Authorization": ""}).status_code == 401
|
|
assert client.post("/api/v1/ingest", json=payload(), headers={"Authorization": "Bearer nope"}).status_code == 401
|
|
|
|
|
|
def test_running_then_finished_merges_into_one_run(client):
|
|
started = iso(-timedelta(hours=1))
|
|
r = client.post("/api/v1/ingest", json=payload({"status": "running", "started_at": started}))
|
|
assert r.status_code == 200, r.text
|
|
job_id = r.json()["job_id"]
|
|
|
|
job = client.get("/api/v1/jobs").json()["jobs"][0]
|
|
assert job["status"] == "running"
|
|
assert job["running_since"] is not None
|
|
assert job["last_run_at"] is None # nothing has finished yet
|
|
|
|
finished = iso(-timedelta(minutes=10))
|
|
client.post("/api/v1/ingest", json=payload(
|
|
{"status": "warning", "started_at": started, "finished_at": finished, "message": "1 file skipped"}))
|
|
|
|
detail = client.get(f"/api/v1/jobs/{job_id}").json()
|
|
assert detail["status"] == "warning"
|
|
assert detail["running_since"] is None
|
|
assert detail["last_run_at"] == to_db(utcnow().fromisoformat(finished))
|
|
assert detail["duration_seconds"] == 50 * 60
|
|
assert detail["message"] == "1 file skipped"
|
|
assert len(client.get(f"/api/v1/jobs/{job_id}/runs").json()["runs"]) == 1
|
|
|
|
|
|
def test_late_running_report_does_not_reopen_finished_run(client):
|
|
started = iso(-timedelta(hours=1))
|
|
client.post("/api/v1/ingest", json=payload({"status": "success", "started_at": started, "finished_at": iso(-timedelta(minutes=5))}))
|
|
client.post("/api/v1/ingest", json=payload({"status": "running", "started_at": started}))
|
|
assert client.get("/api/v1/jobs").json()["jobs"][0]["status"] == "success"
|
|
|
|
|
|
def test_last_run_is_previous_finish_while_new_run_in_progress(client):
|
|
prev_end = iso(-timedelta(hours=23))
|
|
client.post("/api/v1/ingest", json=payload({"status": "failed", "started_at": iso(-timedelta(hours=24)), "finished_at": prev_end, "message": "boom"}))
|
|
client.post("/api/v1/ingest", json=payload({"status": "running", "started_at": iso(-timedelta(minutes=5))}))
|
|
job = client.get("/api/v1/jobs").json()["jobs"][0]
|
|
assert job["status"] == "running"
|
|
assert job["last_run_at"] == to_db(utcnow().fromisoformat(prev_end))
|
|
detail = client.get(f"/api/v1/jobs/{job['id']}").json()
|
|
assert detail["last_status"] == "failed"
|
|
assert detail["message"] == "boom"
|
|
|
|
|
|
def test_next_run_at_only_changes_when_sent(client):
|
|
nxt = iso(timedelta(hours=5))
|
|
client.post("/api/v1/ingest", json=payload(next_run_at=nxt))
|
|
client.post("/api/v1/ingest", json=payload()) # omitted → keep
|
|
assert client.get("/api/v1/jobs").json()["jobs"][0]["next_run_at"] == to_db(utcnow().fromisoformat(nxt))
|
|
client.post("/api/v1/ingest", json=payload(next_run_at=None)) # explicit null → clear
|
|
assert client.get("/api/v1/jobs").json()["jobs"][0]["next_run_at"] is None
|
|
|
|
|
|
def test_history_window_and_run_detail(client):
|
|
for days_ago in (40, 20, 1):
|
|
start = iso(-timedelta(days=days_ago))
|
|
client.post("/api/v1/ingest", json=payload({"status": "success", "started_at": start,
|
|
"finished_at": start, "message": f"{days_ago}d"}))
|
|
job_id = client.get("/api/v1/jobs").json()["jobs"][0]["id"]
|
|
runs = client.get(f"/api/v1/jobs/{job_id}/runs").json()["runs"]
|
|
assert len(runs) == 2 # the 40-day-old run is outside the 30-day window
|
|
assert "message" not in runs[0]
|
|
run = client.get(f"/api/v1/runs/{runs[0]['id']}").json()
|
|
assert run["message"] == "1d"
|
|
assert client.get(f"/api/v1/jobs/{job_id}/runs?days=31").status_code == 422
|
|
|
|
|
|
@pytest.mark.parametrize("run", [
|
|
{"status": "running", "started_at": iso(-timedelta(hours=1)), "finished_at": iso(timedelta(0))},
|
|
{"status": "success", "started_at": iso(timedelta(0)), "finished_at": iso(-timedelta(hours=1))},
|
|
{"status": "success", "started_at": "2026-01-01T00:00:00"}, # no timezone
|
|
{"status": "done", "started_at": iso(timedelta(0))},
|
|
])
|
|
def test_rejects_bad_runs(client, run):
|
|
assert client.post("/api/v1/ingest", json=payload(run)).status_code == 422
|
|
|
|
|
|
def test_tenant_name_is_case_insensitive(client):
|
|
client.post("/api/v1/ingest", json=payload())
|
|
body = payload()
|
|
body["tenant"] = "ACME CORP"
|
|
client.post("/api/v1/ingest", json=body)
|
|
assert len(client.get("/api/v1/jobs").json()["jobs"]) == 1
|
|
|
|
|
|
def test_me_reports_signed_in_user_from_proxy_headers(client):
|
|
assert client.get("/api/v1/me").json() == {"email": None, "name": None}
|
|
r = client.get("/api/v1/me", headers={"X-Forwarded-Email": "[email protected]", "X-Forwarded-User": "abc-123"})
|
|
assert r.json() == {"email": "[email protected]", "name": "abc-123"}
|