Initial scaffold: FastAPI backend + Vue 3 frontend + Docker setup
Backend: config/db/security/logging core, SQLAlchemy models (Server, Credential, UpdateJob, UpdateLog, AuditLog, User), services (winrm, ssh, cau, audit, job_runner), REST API (auth, servers, updates, audit), Socket.io WebSocket layer. Frontend: Vue 3 + TS + Pinia + Tailwind, Views (Dashboard, Servers, Updates, Audit, Login), axios + socket.io-client, nginx prod config.
This commit is contained in:
@@ -0,0 +1,100 @@
|
||||
"""Security helpers: Fernet credential encryption, JWT issue/verify, password hashing."""
|
||||
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import bcrypt
|
||||
from cryptography.fernet import Fernet, InvalidToken
|
||||
from jose import JWTError, jwt
|
||||
|
||||
from app.core.config import get_settings
|
||||
from app.core.exceptions import CredentialDecryptionError, InvalidTokenError
|
||||
|
||||
settings = get_settings()
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fernet encryption for stored credentials
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_fernet: Fernet | None = None
|
||||
|
||||
|
||||
def _get_fernet() -> Fernet:
|
||||
global _fernet
|
||||
if _fernet is None:
|
||||
key = settings.encryption_key
|
||||
if not key:
|
||||
# Dev fallback: derive a valid fernet key from SECRET_KEY
|
||||
import base64
|
||||
import hashlib
|
||||
|
||||
key = base64.urlsafe_b64encode(
|
||||
hashlib.sha256(settings.secret_key.encode()).digest()
|
||||
).decode()
|
||||
_fernet = Fernet(key.encode() if isinstance(key, str) else key)
|
||||
return _fernet
|
||||
|
||||
|
||||
def encrypt(plaintext: str) -> str:
|
||||
"""Encrypt a secret for at-rest storage."""
|
||||
return _get_fernet().encrypt(plaintext.encode()).decode()
|
||||
|
||||
|
||||
def decrypt(token: str) -> str:
|
||||
"""Decrypt a stored secret. Raises CredentialDecryptionError on failure."""
|
||||
try:
|
||||
return _get_fernet().decrypt(token.encode()).decode()
|
||||
except InvalidToken as exc:
|
||||
raise CredentialDecryptionError("Stored credential cannot be decrypted") from exc
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Password hashing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# JWT (RS256 with key files, HS256 fallback for dev without keys)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user