B0rbor4d
cc4c3fcecb
- Backend: Customer/Satellite Models, customer_id auf Server/Job/Audit - Satellite-API: heartbeat, poll (atomares Claiming), logs, result, scan-result, health-report - Auth via X-Api-Key (SHA-256 gehasht) - Job-Queue: pending/claimed/running/success/failed + Stale-Janitor - Batch-Trigger: ein Job pro Server, Satellite arbeitet sequenziell ab - Credentials bleiben lokal: nur symbolische credential_ref zentral - Neues Paket satellite/: Pull-Loop, WinRM/SSH/CAU/Scanner, PyInstaller-tauglich - Frontend: Kunden-Switcher, Satelliten-View, Polling statt WebSocket - Entfernt: WebSocket/Socket.io, Redis, zentrale Credentials, JobRunner - Docs: README/AGENTS/PROMPT auf neue Architektur aktualisiert
59 lines
2.0 KiB
Python
59 lines
2.0 KiB
Python
"""Security helpers: JWT issue/verify, password hashing.
|
|
|
|
No credential encryption here - target-system credentials live exclusively
|
|
on the satellites, never in the central database.
|
|
"""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import bcrypt
|
|
from jose import JWTError, jwt
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.exceptions import InvalidTokenError
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
# bcrypt hard limit: 72 bytes
|
|
return bcrypt.hashpw(password.encode()[:72], bcrypt.gensalt()).decode()
|
|
|
|
|
|
def verify_password(plain: str, hashed: str) -> bool:
|
|
try:
|
|
return bcrypt.checkpw(plain.encode()[:72], hashed.encode())
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _read_key(path: str) -> str | None:
|
|
p = Path(path)
|
|
return p.read_text() if p.exists() else None
|
|
|
|
|
|
def create_access_token(subject: str, extra_claims: dict[str, Any] | None = None) -> str:
|
|
expire = datetime.now(UTC) + timedelta(minutes=settings.jwt_access_token_expire_minutes)
|
|
claims: dict[str, Any] = {"sub": subject, "exp": expire, "type": "access"}
|
|
if extra_claims:
|
|
claims.update(extra_claims)
|
|
|
|
private_key = _read_key(settings.jwt_private_key_path)
|
|
if private_key and settings.jwt_algorithm == "RS256":
|
|
return jwt.encode(claims, private_key, algorithm="RS256")
|
|
# Dev fallback when no key pair exists yet
|
|
return jwt.encode(claims, settings.secret_key, algorithm="HS256")
|
|
|
|
|
|
def decode_token(token: str) -> dict[str, Any]:
|
|
"""Decode and validate a JWT. Raises InvalidTokenError on failure."""
|
|
try:
|
|
public_key = _read_key(settings.jwt_public_key_path)
|
|
if public_key and settings.jwt_algorithm == "RS256":
|
|
return jwt.decode(token, public_key, algorithms=["RS256"]) # type: ignore[no-any-return]
|
|
return jwt.decode(token, settings.secret_key, algorithms=["HS256"]) # type: ignore[no-any-return]
|
|
except JWTError as exc:
|
|
raise InvalidTokenError("Token is invalid or expired") from exc
|