inecs
f8d3cbdd59
Add the FastAPI app, PostgreSQL migrations, Docker/Helm packaging, API contracts, docs, client examples, and the unit/integration/compatibility test suite for local client and tooling labs without a real vCenter.
155 lines
4.9 KiB
Python
155 lines
4.9 KiB
Python
"""Session IDs compatible with vmware-api-session-id header."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from fastapi import Depends, Request
|
|
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
|
|
|
from app.db.pool import AsyncpgDatabase, Database
|
|
from app.dependencies import get_database
|
|
from app.security.auth import hash_secret, verify_secret
|
|
from app.vsphere.errors import unauthenticated
|
|
|
|
SESSION_TTL = timedelta(hours=2)
|
|
SESSION_HEADER = "vmware-api-session-id"
|
|
DEFAULT_USER = "administrator@vsphere.local"
|
|
DEFAULT_PASSWORD = "VMware1!"
|
|
|
|
_basic = HTTPBasic(auto_error=False)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SessionInfo:
|
|
id: str
|
|
username: str
|
|
roles: tuple[str, ...] = field(default_factory=tuple)
|
|
|
|
|
|
def _pool(database: Database) -> Any:
|
|
return database.pool # type: ignore[attr-defined]
|
|
|
|
|
|
async def ensure_default_credentials(database: Database) -> None:
|
|
"""Idempotently insert lab SSO credentials (full set via seed preferred)."""
|
|
|
|
from app.vsphere.profiles import lab_credentials
|
|
|
|
pool = _pool(database)
|
|
async with pool.acquire() as conn:
|
|
for cred in lab_credentials():
|
|
await conn.execute(
|
|
"""
|
|
INSERT INTO vsphere_credentials (username, password_hash, roles)
|
|
VALUES ($1, $2, $3)
|
|
ON CONFLICT (username) DO UPDATE SET
|
|
password_hash = EXCLUDED.password_hash,
|
|
roles = EXCLUDED.roles
|
|
""",
|
|
cred.username,
|
|
hash_secret(cred.password),
|
|
list(cred.roles),
|
|
)
|
|
|
|
|
|
async def verify_password(database: Database, username: str, password: str) -> bool:
|
|
pool = _pool(database)
|
|
async with pool.acquire() as conn:
|
|
row = await conn.fetchrow(
|
|
"SELECT password_hash FROM vsphere_credentials WHERE username = $1",
|
|
username,
|
|
)
|
|
if row is None:
|
|
return username == DEFAULT_USER and password == DEFAULT_PASSWORD
|
|
return verify_secret(password, str(row["password_hash"]))
|
|
|
|
|
|
async def create_session(database: Database, username: str) -> str:
|
|
session_id = secrets.token_hex(16)
|
|
expires = datetime.now(UTC) + SESSION_TTL
|
|
pool = _pool(database)
|
|
async with pool.acquire() as conn:
|
|
await conn.execute(
|
|
"""
|
|
INSERT INTO vsphere_sessions (id, username, expires_at)
|
|
VALUES ($1, $2, $3)
|
|
""",
|
|
session_id,
|
|
username,
|
|
expires,
|
|
)
|
|
return session_id
|
|
|
|
|
|
async def delete_session(database: Database, session_id: str) -> None:
|
|
pool = _pool(database)
|
|
async with pool.acquire() as conn:
|
|
await conn.execute("DELETE FROM vsphere_sessions WHERE id = $1", session_id)
|
|
|
|
|
|
async def _roles_for(database: Database, username: str) -> tuple[str, ...]:
|
|
pool = _pool(database)
|
|
async with pool.acquire() as conn:
|
|
row = await conn.fetchrow(
|
|
"SELECT roles FROM vsphere_credentials WHERE username = $1",
|
|
username,
|
|
)
|
|
if row is None:
|
|
return ("Administrator",) if username == DEFAULT_USER else ("ReadOnly",)
|
|
roles = tuple(str(r) for r in (row["roles"] or []))
|
|
return roles or ("ReadOnly",)
|
|
|
|
|
|
async def lookup_session(database: Database, session_id: str) -> SessionInfo | None:
|
|
pool = _pool(database)
|
|
async with pool.acquire() as conn:
|
|
row = await conn.fetchrow(
|
|
"""
|
|
SELECT id, username, expires_at FROM vsphere_sessions
|
|
WHERE id = $1
|
|
""",
|
|
session_id,
|
|
)
|
|
if row is None:
|
|
return None
|
|
if row["expires_at"] <= datetime.now(UTC):
|
|
await conn.execute("DELETE FROM vsphere_sessions WHERE id = $1", session_id)
|
|
return None
|
|
await conn.execute(
|
|
"UPDATE vsphere_sessions SET expires_at = $2 WHERE id = $1",
|
|
session_id,
|
|
datetime.now(UTC) + SESSION_TTL,
|
|
)
|
|
username = str(row["username"])
|
|
roles = await _roles_for(database, username)
|
|
return SessionInfo(id=str(row["id"]), username=username, roles=roles)
|
|
|
|
|
|
async def require_session(
|
|
request: Request,
|
|
database: Database = Depends(get_database),
|
|
) -> SessionInfo:
|
|
session_id = request.headers.get(SESSION_HEADER) or request.cookies.get(SESSION_HEADER)
|
|
if not session_id:
|
|
raise unauthenticated()
|
|
info = await lookup_session(database, session_id)
|
|
if info is None:
|
|
raise unauthenticated("Invalid or expired session")
|
|
return info
|
|
|
|
|
|
async def optional_basic(
|
|
credentials: HTTPBasicCredentials | None = Depends(_basic),
|
|
) -> HTTPBasicCredentials | None:
|
|
return credentials
|
|
|
|
|
|
def as_asyncpg(database: Database) -> AsyncpgDatabase:
|
|
if not isinstance(database, AsyncpgDatabase):
|
|
raise TypeError("vsphere routes require AsyncpgDatabase")
|
|
return database
|