From 8471bcdbacab39f3df1c591dd14d2ece2cb74431 Mon Sep 17 00:00:00 2001 From: Sergey Antropoff Date: Mon, 13 Jul 2026 01:28:00 +0300 Subject: [PATCH] feat: authenticate API tokens and enforce route ACLs --- README.md | 8 ++ app/api/registry.py | 104 ++++++++++++++++++++----- app/simulation/seed.py | 40 ++++++++++ docs/compatibility-0.1.0.md | 4 +- tests/compatibility/test_proxmoxer.py | 25 +++++- tests/unit/test_api_auth_boundary.py | 105 ++++++++++++++++++++++++++ 6 files changed, 267 insertions(+), 19 deletions(-) create mode 100644 tests/unit/test_api_auth_boundary.py diff --git a/README.md b/README.md index 5e24fbd..a1b8207 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,14 @@ print(proxmox.version.get()) print(proxmox.nodes("pve1").qemu.get()) ``` +The deterministic seed also provides hashed development API tokens. For token +authentication use proxmoxer `token_name="automation"` and +`token_value="automation-secret"` with user `root@pam`. The readonly +`auditor@pve!readonly` token proves privilege separation: authenticated reads +work, while QEMU power operations return 403. API-token requests do not require +CSRF; ticket-authenticated mutations still do. These are disposable local test +credentials only. + Run the external-client smoke flow against the Compose network with `PROXMOXER_HOST=tls-gateway`, `PROXMOXER_PORT=8443`, and pytest marker `compatibility`. It covers login, reads, CSRF-protected mutation, and UPID task diff --git a/app/api/registry.py b/app/api/registry.py index 5dc7cfc..a64967a 100644 --- a/app/api/registry.py +++ b/app/api/registry.py @@ -13,7 +13,9 @@ from fastapi.responses import JSONResponse from app.api.errors import ApiError, ContractValidationError from app.config import Settings from app.contracts.model import Method, Schema, Snapshot -from app.security.auth import verify_csrf, verify_ticket +from app.db.pool import AsyncpgDatabase +from app.security.acl import AclEntry, authorize, requirement_from_contract +from app.security.auth import parse_api_token, verify_csrf, verify_secret, verify_ticket Handler = Callable[[Request, dict[str, Any]], Awaitable[Any]] FallbackMode = Literal["error", "schema-default", "fixture"] @@ -81,9 +83,9 @@ def _endpoint( fallback: FallbackMode, ) -> Callable[[Request], Awaitable[JSONResponse]]: async def dispatch(request: Request) -> JSONResponse: - _authenticate(request, semantic_path) - handler = handlers.get(semantic_path, method.verb) inputs = await _parse_inputs(request, method) + await _authenticate(request, semantic_path, method, inputs) + handler = handlers.get(semantic_path, method.verb) if handler is not None: data = await handler(request, inputs) elif fallback == "schema-default": @@ -108,22 +110,90 @@ def _endpoint( return dispatch -def _authenticate(request: Request, semantic_path: str) -> None: +async def _authenticate( + request: Request, semantic_path: str, method: Method, inputs: dict[str, Any] +) -> None: if semantic_path in {"/version", "/access/ticket"}: return - ticket = request.cookies.get("PVEAuthCookie") - if ticket is None: - raise ApiError(401, "authentication required") - settings = cast(Settings, request.app.state.settings) - key = settings.ticket_signing_key.get_secret_value().encode() - try: - verify_ticket(ticket, key) - except ValueError as error: - raise ApiError(401, "authentication failure") from error - if request.method not in {"GET", "HEAD", "OPTIONS"}: - token = request.headers.get("CSRFPreventionToken", "") - if not verify_csrf(ticket, token, key): - raise ApiError(403, "invalid CSRF prevention token") + authorization = request.headers.get("Authorization", "") + token_privileges: frozenset[str] | None = None + principal: str + if authorization.startswith("PVEAPIToken="): + database = cast(AsyncpgDatabase, request.app.state.database) + try: + parsed_token = parse_api_token(authorization) + except ValueError as error: + raise ApiError(401, "authentication failure") from error + row = await database.pool.fetchrow( + """SELECT p.name, t.secret_hash, t.privileges + FROM api_tokens t JOIN principals p ON p.id=t.principal_id + WHERE p.name=$1 AND t.token_id=$2 + AND (t.expires_at IS NULL OR t.expires_at > now())""", + parsed_token.principal, + parsed_token.token_id, + ) + if row is None or not verify_secret(parsed_token.secret, str(row["secret_hash"])): + raise ApiError(401, "authentication failure") + principal = str(row["name"]) + token_privileges = frozenset(str(item) for item in row["privileges"]) + else: + ticket = request.cookies.get("PVEAuthCookie") + if ticket is None: + raise ApiError(401, "authentication required") + settings = cast(Settings, request.app.state.settings) + key = settings.ticket_signing_key.get_secret_value().encode() + try: + claims = verify_ticket(ticket, key) + except ValueError as error: + raise ApiError(401, "authentication failure") from error + principal = claims.principal + if request.method not in {"GET", "HEAD", "OPTIONS"}: + csrf_value = request.headers.get("CSRFPreventionToken", "") + if not verify_csrf(ticket, csrf_value, key): + raise ApiError(403, "invalid CSRF prevention token") + request.state.principal = principal + if principal == "root@pam" and token_privileges is None: + return + database = cast(AsyncpgDatabase, request.app.state.database) + await _authorize(database, principal, token_privileges, method, inputs) + + +async def _authorize( + database: AsyncpgDatabase, + principal: str, + token_privileges: frozenset[str] | None, + method: Method, + inputs: dict[str, Any], +) -> None: + values = cast(dict[str, Any], inputs["values"]) + requirement = requirement_from_contract( + method.permissions, {name: str(value) for name, value in values.items()} + ) + if requirement is None: + return + rows = await database.pool.fetch( + """SELECT a.path, a.propagate, r.privileges + FROM acl_entries a JOIN roles r ON r.name=a.role_name + JOIN principals p ON p.id=a.principal_id WHERE p.name=$1""", + principal, + ) + entries = tuple( + AclEntry( + principal, + str(row["path"]), + frozenset(str(item) for item in row["privileges"]), + bool(row["propagate"]), + ) + for row in rows + ) + if not authorize( + principal, + requirement.path, + requirement.privileges, + entries, + token_privileges=token_privileges, + ): + raise ApiError(403, "permission check failed") async def _parse_inputs(request: Request, method: Method) -> dict[str, Any]: diff --git a/app/simulation/seed.py b/app/simulation/seed.py index 1a3cb1b..1959cf1 100644 --- a/app/simulation/seed.py +++ b/app/simulation/seed.py @@ -332,6 +332,46 @@ async def apply_seed(connection: Connection, profile: SeedProfile) -> None: stable_id("principal:root@pam"), hash_secret("secret", salt=b"pve-simulator-v1"), ) + await connection.execute( + """INSERT INTO api_tokens(principal_id, token_id, secret_hash, privileges) + VALUES($1, 'automation', $2, $3) + ON CONFLICT (principal_id, token_id) DO UPDATE + SET secret_hash=EXCLUDED.secret_hash, privileges=EXCLUDED.privileges""", + stable_id("principal:root@pam"), + hash_secret("automation-secret", salt=b"pve-token-seed-v1"), + ["VM.Audit", "VM.PowerMgmt", "Sys.Audit"], + ) + auditor_id = stable_id("principal:auditor@pve") + await connection.execute( + """INSERT INTO principals(id, name, password_hash, realm_name) + VALUES($1, 'auditor@pve', $2, 'pve') + ON CONFLICT (name) DO UPDATE SET password_hash=EXCLUDED.password_hash, + realm_name=EXCLUDED.realm_name""", + auditor_id, + hash_secret("auditor-secret", salt=b"pve-auditor-v1"), + ) + await connection.execute( + """INSERT INTO roles(name, privileges) + VALUES('PVEAuditor', $1) + ON CONFLICT (name) DO UPDATE SET privileges=EXCLUDED.privileges""", + ["Sys.Audit", "VM.Audit"], + ) + await connection.execute( + """INSERT INTO acl_entries(principal_id, role_name, path, propagate) + VALUES($1, 'PVEAuditor', '/', true) + ON CONFLICT (principal_id, role_name, path) DO UPDATE + SET propagate=EXCLUDED.propagate""", + auditor_id, + ) + await connection.execute( + """INSERT INTO api_tokens(principal_id, token_id, secret_hash, privileges) + VALUES($1, 'readonly', $2, $3) + ON CONFLICT (principal_id, token_id) DO UPDATE + SET secret_hash=EXCLUDED.secret_hash, privileges=EXCLUDED.privileges""", + auditor_id, + hash_secret("readonly-secret", salt=b"pve-readonly-v1"), + ["Sys.Audit", "VM.Audit"], + ) async def seed_url( diff --git a/docs/compatibility-0.1.0.md b/docs/compatibility-0.1.0.md index cb21664..7cf186d 100644 --- a/docs/compatibility-0.1.0.md +++ b/docs/compatibility-0.1.0.md @@ -24,7 +24,9 @@ worker tests; a single smoke run chooses the transition valid for current state. - Core: version, ticket login, node list/status, and cluster resources. - QEMU: list, configuration, current status, start, and stop. - Tasks: node task list, status, and append-only log. -- Authentication: ticket cookie and ticket-bound CSRF validation for mutations. +- Authentication: ticket cookie and ticket-bound CSRF validation for mutations, + plus hashed API-token authentication without CSRF and token privilege + separation at the contract-derived ACL boundary. - Persistence: PostgreSQL resources, durable leased tasks, and deterministic `small` seed data. diff --git a/tests/compatibility/test_proxmoxer.py b/tests/compatibility/test_proxmoxer.py index 20d49fd..5bc789e 100644 --- a/tests/compatibility/test_proxmoxer.py +++ b/tests/compatibility/test_proxmoxer.py @@ -4,7 +4,7 @@ import os from threading import Event import pytest -from proxmoxer import ProxmoxAPI # type: ignore[import-untyped] +from proxmoxer import ProxmoxAPI, ResourceException # type: ignore[import-untyped] pytestmark = [ pytest.mark.compatibility, @@ -25,6 +25,29 @@ def test_proxmoxer_read_and_qemu_task_flow() -> None: assert any(node["node"] == "pve1" for node in proxmox.nodes.get()) assert any(vm["vmid"] == 101 for vm in proxmox.nodes("pve1").qemu.get()) + token_api = ProxmoxAPI( + os.environ["PROXMOXER_HOST"], + port=int(os.getenv("PROXMOXER_PORT", "8007")), + user="root@pam", + token_name=os.getenv("PROXMOXER_TOKEN_NAME", "automation"), + token_value=os.getenv("PROXMOXER_TOKEN_SECRET", "automation-secret"), + verify_ssl=False, + ) + assert any(node["node"] == "pve1" for node in token_api.nodes.get()) + + readonly_api = ProxmoxAPI( + os.environ["PROXMOXER_HOST"], + port=int(os.getenv("PROXMOXER_PORT", "8007")), + user="auditor@pve", + token_name=os.getenv("PROXMOXER_READONLY_TOKEN_NAME", "readonly"), + token_value=os.getenv("PROXMOXER_READONLY_TOKEN_SECRET", "readonly-secret"), + verify_ssl=False, + ) + assert readonly_api.nodes.get() + with pytest.raises(ResourceException) as denied: + readonly_api.nodes("pve1").qemu("101").status.start.post() + assert denied.value.status_code == 403 + if os.getenv("PROXMOXER_MUTATION_TEST") == "1": status = proxmox.nodes("pve1").qemu("101").status.current.get() operation = "start" if status["status"] == "stopped" else "stop" diff --git a/tests/unit/test_api_auth_boundary.py b/tests/unit/test_api_auth_boundary.py new file mode 100644 index 0000000..7fe3199 --- /dev/null +++ b/tests/unit/test_api_auth_boundary.py @@ -0,0 +1,105 @@ +"""HTTP-boundary API-token and contract permission tests.""" + +from typing import Any, cast + +import pytest +from fastapi import FastAPI, Request + +from app.api.errors import ApiError +from app.api.registry import _authenticate +from app.config import Settings +from app.contracts.model import Method, Permissions, Schema +from app.db.pool import AsyncpgDatabase +from app.security.auth import hash_secret + + +class FakePool: + def __init__(self, secret: str, token_privileges: list[str]) -> None: + self.secret_hash = hash_secret(secret, salt=b"boundary-token-v1") + self.token_privileges = token_privileges + + async def fetchrow(self, _query: str, principal: str, token_id: str) -> dict[str, Any] | None: + if principal != "operator@pve" or token_id != "api": + return None + return { + "name": principal, + "secret_hash": self.secret_hash, + "privileges": self.token_privileges, + } + + async def fetch(self, _query: str, principal: str) -> list[dict[str, Any]]: + return [ + { + "path": "/vms", + "propagate": True, + "privileges": ["VM.Audit", "VM.PowerMgmt"], + "principal": principal, + } + ] + + +class FakeDatabase: + def __init__(self, pool: FakePool) -> None: + self.pool = pool + + +def token_request(secret: str, token_privileges: list[str]) -> Request: + app = FastAPI() + app.state.settings = Settings() + app.state.database = cast(AsyncpgDatabase, FakeDatabase(FakePool("valid", token_privileges))) + return Request( + { + "type": "http", + "app": app, + "method": "POST", + "path": "/api2/json/nodes/pve1/qemu/101/status/start", + "headers": [(b"authorization", f"PVEAPIToken=operator@pve!api={secret}".encode())], + "query_string": b"", + "server": ("test", 80), + "client": ("test", 123), + "scheme": "http", + } + ) + + +def power_method() -> Method: + return Method( + verb="POST", + name="start", + returns=Schema(type="string"), + permissions=Permissions(expression={"check": ["perm", "/vms/{vmid}", ["VM.PowerMgmt"]]}), + checksum="1" * 64, + ) + + +async def test_api_token_skips_csrf_but_honors_separated_privileges() -> None: + allowed = token_request("valid", ["VM.PowerMgmt"]) + await _authenticate( + allowed, + "/nodes/{node}/qemu/{vmid}/status/start", + power_method(), + {"values": {"node": "pve1", "vmid": 101}}, + ) + assert allowed.state.principal == "operator@pve" + + denied = token_request("valid", ["VM.Audit"]) + with pytest.raises(ApiError) as error: + await _authenticate( + denied, + "/nodes/{node}/qemu/{vmid}/status/start", + power_method(), + {"values": {"node": "pve1", "vmid": 101}}, + ) + assert error.value.status_code == 403 + + +async def test_api_token_rejects_unknown_or_wrong_secret() -> None: + request = token_request("wrong", ["VM.PowerMgmt"]) + with pytest.raises(ApiError) as error: + await _authenticate( + request, + "/nodes/{node}/qemu/{vmid}/status/start", + power_method(), + {"values": {"node": "pve1", "vmid": 101}}, + ) + assert error.value.status_code == 401