Initial commit: AegisSight-Monitor-Verwaltung

src/auth.py

@@ -0,0 +1,61 @@
+"""Passwort-basierte Authentifizierung fuer das Verwaltungsportal."""
+from datetime import datetime, timedelta, timezone
+from jose import jwt, JWTError
+import bcrypt as _bcrypt
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from config import JWT_SECRET, JWT_ALGORITHM, JWT_EXPIRE_HOURS
+
+security = HTTPBearer()
+
+JWT_ISSUER = "intelsight-portal"
+JWT_AUDIENCE = "intelsight-portal"
+
+
+def hash_password(password: str) -> str:
+    return _bcrypt.hashpw(password.encode("utf-8"), _bcrypt.gensalt()).decode("utf-8")
+
+
+def verify_password(password: str, password_hash: str) -> bool:
+    return _bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8"))
+
+
+def create_token(admin_id: int, username: str) -> str:
+    now = datetime.now(timezone.utc)
+    expire = now + timedelta(hours=JWT_EXPIRE_HOURS)
+    payload = {
+        "sub": str(admin_id),
+        "username": username,
+        "role": "portal_admin",
+        "iss": JWT_ISSUER,
+        "aud": JWT_AUDIENCE,
+        "iat": now,
+        "exp": expire,
+    }
+    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
+
+
+def decode_token(token: str) -> dict:
+    try:
+        return jwt.decode(
+            token,
+            JWT_SECRET,
+            algorithms=[JWT_ALGORITHM],
+            issuer=JWT_ISSUER,
+            audience=JWT_AUDIENCE,
+        )
+    except JWTError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token ungueltig oder abgelaufen",
+        )
+
+
+async def get_current_admin(
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+) -> dict:
+    payload = decode_token(credentials.credentials)
+    return {
+        "id": int(payload["sub"]),
+        "username": payload["username"],
+    }