Claude Code
92bed208e0
ARCHITECTURE ============ Backend: FastAPI + uvicorn (port 8000) - JWT authentication with PAM system users - ZFS CLI wrapper with caching (30-60s TTL) - WebSocket pool status broadcaster (30s interval) - Services: auth, zfs_runner, file_manager, shares, identities, system_info - Routers: pools, datasets, snapshots, shares, identities, navigator, system Frontend: Next.js 15 + TypeScript (static export) - Incremental Static Regeneration (ISR) for weak hardware - Type-safe API client (lib/api.ts) - Dark mode + custom Tailwind theme - Pages: Dashboard, Login, Snapshots, Datasets, Shares, etc. DEPLOYMENT ========== Test Target: 192.168.1.179:8090 (Debian LXC) Production: 10.66.120.3:9090 (Raspberry Pi 4GB ARM64) Updater: Automated Gitea-based deployment (update-test.sh, update-pi.sh) FEATURES COMPLETED ================== Phase 3a: Dashboard Quick Stats (System, CPU, Memory, Storage) - Real-time stats with color-coded progress bars - Responsive grid layout (mobile: 1, tablet: 2, desktop: 4 columns) - ISR-optimized for fast loads on weak hardware REBRANDING ========== Renamed throughout: - Project: 'ZFS Manager' → 'ZMB Webui' - Services: 'zfs-manager' → 'zmb-webui' - Systemd units: zfs-manager-backend → zmb-webui-backend - Configuration files and documentation Co-Authored-By: Patrick <patrick@perlbach24.de>
90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
"""
|
|
JWT Authentication Service
|
|
Handles user login via PAM (Linux system users), token generation, and verification
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
|
|
from jose import JWTError, jwt
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# JWT Configuration
|
|
SECRET_KEY = os.environ.get("ZFS_SECRET_KEY", "your-secret-key-change-in-production")
|
|
ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_HOURS = 8
|
|
|
|
# Try to import PAM for system authentication
|
|
try:
|
|
import pam
|
|
PAM_AVAILABLE = True
|
|
except ImportError:
|
|
PAM_AVAILABLE = False
|
|
logger.warning("python-pam not installed, PAM authentication unavailable")
|
|
|
|
|
|
class AuthService:
|
|
def __init__(self):
|
|
"""Initialize auth service with PAM (Linux system users)"""
|
|
if PAM_AVAILABLE:
|
|
logger.info("Using PAM authentication (Linux system users)")
|
|
else:
|
|
logger.error("PAM not available - install python-pam for authentication")
|
|
|
|
def authenticate_user(self, username: str, password: str) -> Optional[dict]:
|
|
"""
|
|
Authenticate user via PAM (Linux system users like 'pi', 'root')
|
|
Returns user data if valid, None otherwise
|
|
"""
|
|
if not PAM_AVAILABLE:
|
|
logger.error("PAM not available")
|
|
return None
|
|
|
|
try:
|
|
p = pam.pam()
|
|
if p.authenticate(username, password):
|
|
logger.info(f"User {username} authenticated via PAM")
|
|
return {
|
|
"username": username,
|
|
"source": "pam"
|
|
}
|
|
else:
|
|
logger.warning(f"PAM authentication failed for user {username}: {p.reason}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"PAM authentication error: {e}")
|
|
return None
|
|
|
|
def create_access_token(self, username: str, expires_delta: Optional[timedelta] = None) -> str:
|
|
"""Create JWT access token"""
|
|
if expires_delta is None:
|
|
expires_delta = timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS)
|
|
|
|
expire = datetime.utcnow() + expires_delta
|
|
to_encode = {"sub": username, "exp": expire}
|
|
|
|
try:
|
|
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
|
return encoded_jwt
|
|
except Exception as e:
|
|
logger.error(f"Failed to create token: {e}")
|
|
raise
|
|
|
|
def verify_token(self, token: str) -> Optional[str]:
|
|
"""Verify JWT token and return username"""
|
|
try:
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
username: str = payload.get("sub")
|
|
if username is None:
|
|
return None
|
|
return username
|
|
except JWTError:
|
|
return None
|
|
|
|
|
|
# Global instance
|
|
auth_service = AuthService()
|