Gan, Jimmy
422 lines
17 KiB
Python
422 lines
17 KiB
Python
from datetime import timedelta
|
|
import asyncio
|
|
import time
|
|
import json
|
|
import httpx
|
|
from fastapi import APIRouter, Depends, HTTPException, status, Request
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel
|
|
from slowapi import Limiter
|
|
from slowapi.util import get_remote_address
|
|
import jwt
|
|
import config
|
|
import auth
|
|
import logging
|
|
import pyotp
|
|
from webauthn import generate_registration_options, verify_registration_response, generate_authentication_options, verify_authentication_response
|
|
from webauthn.helpers.structs import PublicKeyCredentialDescriptor, UserVerificationRequirement
|
|
from webauthn.helpers import bytes_to_base64url, base64url_to_bytes, options_to_json
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
limiter = Limiter(key_func=get_remote_address)
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
totp_code: str = ""
|
|
|
|
class Token(BaseModel):
|
|
access_token: str
|
|
refresh_token: str
|
|
token_type: str
|
|
user: str
|
|
has_2fa: bool
|
|
role: str = "admin"
|
|
|
|
class RefreshRequest(BaseModel):
|
|
refresh_token: str
|
|
|
|
class Setup2FAResponse(BaseModel):
|
|
secret: str
|
|
uri: str
|
|
|
|
class Verify2FARequest(BaseModel):
|
|
secret: str
|
|
code: str
|
|
|
|
async def send_failed_login_alert(ip_address: str, username: str, reason: str):
|
|
logger.info("Triggering Telegram alert for %s from %s: %s", username, ip_address, reason)
|
|
if not config.TELEGRAM_BOT_TOKEN or not config.TELEGRAM_CHAT_ID:
|
|
logger.warning("Skipping Telegram alert: missing config")
|
|
return
|
|
msg = f"🚨 *Dashboard Security Alert* 🚨\nFailed login attempt detected.\n\n👤 User: `{username}`\n🌐 IP: `{ip_address}`\n❌ Reason: {reason}"
|
|
try:
|
|
client_options = {}
|
|
if config.TELEGRAM_PROXY:
|
|
client_options["proxy"] = config.TELEGRAM_PROXY
|
|
|
|
async with httpx.AsyncClient(**client_options) as client:
|
|
res = await client.post(
|
|
f"https://api.telegram.org/bot{config.TELEGRAM_BOT_TOKEN}/sendMessage",
|
|
data={"chat_id": config.TELEGRAM_CHAT_ID, "text": msg, "parse_mode": "Markdown"},
|
|
timeout=10.0
|
|
)
|
|
logger.info("Telegram alert sent, status: %s", res.status_code)
|
|
except Exception as e:
|
|
logger.warning("Failed to send Telegram alert: %s", e)
|
|
|
|
@router.post("/login", response_model=Token)
|
|
@limiter.limit("5/minute")
|
|
async def login(creds: LoginRequest, request: Request):
|
|
client_ip = request.client.host
|
|
# Verify username/password
|
|
if creds.username != config.ADMIN_USER or not auth.verify_password(creds.password, auth.load_password_hash()):
|
|
asyncio.create_task(send_failed_login_alert(client_ip, creds.username, "Incorrect username or password"))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
# Check 2FA
|
|
totp_secret = auth.load_totp_secret()
|
|
if totp_secret:
|
|
if not creds.totp_code:
|
|
asyncio.create_task(send_failed_login_alert(client_ip, creds.username, "Missing 2FA Code"))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="2FA code required",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
if not auth.verify_totp(creds.totp_code, totp_secret):
|
|
asyncio.create_task(send_failed_login_alert(client_ip, creds.username, "Invalid 2FA Code"))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
# Direct login is admin-only fallback
|
|
access_token = auth.create_access_token(
|
|
data={"sub": creds.username, "role": "admin"},
|
|
expires_delta=timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES),
|
|
)
|
|
refresh_token = auth.create_refresh_token(data={"sub": creds.username, "role": "admin"})
|
|
return {
|
|
"access_token": access_token,
|
|
"refresh_token": refresh_token,
|
|
"token_type": "bearer",
|
|
"user": creds.username,
|
|
"has_2fa": bool(totp_secret),
|
|
"role": "admin",
|
|
}
|
|
|
|
@router.get("/proxy")
|
|
async def proxy_auth(request: Request):
|
|
"""Auto-login when Authelia has already authenticated the user via forward-auth.
|
|
Caddy sets Remote-User and Remote-Groups headers after successful Authelia verification."""
|
|
import config
|
|
from rbac import resolve_role
|
|
if not config.is_trusted_proxy(request.client.host):
|
|
logger.warning(f"Rejected proxy auth attempt from unauthorized IP: {request.client.host}")
|
|
raise HTTPException(status_code=403, detail="Unauthorized proxy source")
|
|
|
|
remote_user = request.headers.get("Remote-User")
|
|
if not remote_user:
|
|
raise HTTPException(status_code=401, detail="No proxy auth header")
|
|
|
|
# Parse groups from Remote-Groups header (comma-separated)
|
|
remote_groups_raw = request.headers.get("Remote-Groups", "")
|
|
groups = [g.strip() for g in remote_groups_raw.split(",") if g.strip()]
|
|
|
|
role = resolve_role(groups)
|
|
if role is None:
|
|
logger.warning(f"User {remote_user} has no dashboard role (groups: {groups})")
|
|
raise HTTPException(status_code=403, detail="User not authorized for dashboard")
|
|
|
|
access_token = auth.create_access_token(
|
|
data={"sub": remote_user, "role": role},
|
|
expires_delta=timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES),
|
|
)
|
|
refresh_token = auth.create_refresh_token(data={"sub": remote_user, "role": role})
|
|
return {
|
|
"access_token": access_token,
|
|
"refresh_token": refresh_token,
|
|
"token_type": "bearer",
|
|
"user": remote_user,
|
|
"has_2fa": True,
|
|
"role": role,
|
|
}
|
|
|
|
@router.get("/me")
|
|
async def read_users_me(current_user = Depends(auth.get_current_user)):
|
|
totp_secret = auth.load_totp_secret()
|
|
return {
|
|
"username": current_user.username,
|
|
"role": current_user.role,
|
|
"pages": current_user.pages,
|
|
"has_2fa": bool(totp_secret),
|
|
}
|
|
|
|
@router.post("/refresh")
|
|
@limiter.limit("10/minute")
|
|
async def refresh_token(req: RefreshRequest, request: Request):
|
|
try:
|
|
payload = jwt.decode(req.refresh_token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
|
|
if payload.get("type") != "refresh":
|
|
raise HTTPException(status_code=401, detail="Invalid token type")
|
|
username = payload.get("sub")
|
|
role = payload.get("role", "admin")
|
|
if username is None:
|
|
raise HTTPException(status_code=401, detail="Invalid user")
|
|
if not auth.verify_token_version(payload.get("tv", -1)):
|
|
raise HTTPException(status_code=401, detail="Token revoked")
|
|
except jwt.PyJWTError:
|
|
raise HTTPException(status_code=401, detail="Invalid refresh token")
|
|
auth.increment_token_version()
|
|
access_token = auth.create_access_token(
|
|
data={"sub": username, "role": role},
|
|
expires_delta=timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES),
|
|
)
|
|
refresh_token = auth.create_refresh_token(data={"sub": username, "role": role})
|
|
return {"access_token": access_token, "refresh_token": refresh_token}
|
|
|
|
@router.post("/setup-2fa/generate", response_model=Setup2FAResponse)
|
|
async def generate_2fa(current_user = Depends(auth.get_current_user)):
|
|
secret = pyotp.random_base32()
|
|
uri = pyotp.totp.TOTP(secret).provisioning_uri(name=current_user.username, issuer_name="NAS Dashboard")
|
|
return {"secret": secret, "uri": uri}
|
|
|
|
@router.post("/setup-2fa/verify")
|
|
async def verify_2fa_setup(request: Verify2FARequest, current_user = Depends(auth.get_current_user)):
|
|
totp = pyotp.TOTP(request.secret)
|
|
if not totp.verify(request.code):
|
|
raise HTTPException(status_code=400, detail="Invalid code")
|
|
|
|
# Save secret
|
|
auth.save_totp_secret(request.secret)
|
|
return {"message": "2FA enabled successfully"}
|
|
|
|
@router.post("/setup-2fa/disable")
|
|
async def disable_2fa(current_user = Depends(auth.get_current_user)):
|
|
auth.save_totp_secret("")
|
|
return {"message": "2FA disabled"}
|
|
|
|
class ChangePasswordRequest(BaseModel):
|
|
current_password: str
|
|
new_password: str
|
|
|
|
@router.post("/change-password")
|
|
@limiter.limit("5/minute")
|
|
async def change_password(req: ChangePasswordRequest, request: Request, current_user = Depends(auth.get_current_user)):
|
|
if not auth.verify_password(req.current_password, auth.load_password_hash()):
|
|
raise HTTPException(status_code=400, detail="Current password is incorrect")
|
|
if len(req.new_password) < 8:
|
|
raise HTTPException(status_code=400, detail="Password must be at least 8 characters")
|
|
auth.save_password_hash(auth.get_password_hash(req.new_password))
|
|
return {"message": "Password changed successfully"}
|
|
|
|
# WebAuthn — in-memory challenge store with TTL
|
|
_challenges = {}
|
|
|
|
def _store_challenge(challenge: bytes):
|
|
_challenges[challenge] = time.time()
|
|
now = time.time()
|
|
expired = [k for k, v in _challenges.items() if now - v > 300]
|
|
for k in expired:
|
|
_challenges.pop(k, None)
|
|
|
|
def _get_challenge(client_data_b64: str) -> bytes:
|
|
if not client_data_b64:
|
|
raise HTTPException(status_code=400, detail="Missing clientDataJSON")
|
|
try:
|
|
data = json.loads(base64url_to_bytes(client_data_b64).decode('utf-8'))
|
|
chal_bytes = base64url_to_bytes(data.get("challenge", ""))
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid clientDataJSON")
|
|
|
|
entry = _challenges.pop(chal_bytes, None)
|
|
if not entry or time.time() - entry > 300:
|
|
raise HTTPException(status_code=400, detail="Challenge expired or invalid")
|
|
return chal_bytes
|
|
|
|
@router.post("/passkey/register/options")
|
|
async def passkey_register_options(current_user = Depends(auth.get_current_user)):
|
|
existing = auth.load_passkey_credentials()
|
|
exclude = [PublicKeyCredentialDescriptor(id=base64url_to_bytes(c["credential_id"])) for c in existing]
|
|
options = generate_registration_options(
|
|
rp_id=config.WEBAUTHN_RP_ID,
|
|
rp_name="NAS Dashboard",
|
|
user_id=current_user.username.encode(),
|
|
user_name=current_user.username,
|
|
user_display_name=current_user.username,
|
|
exclude_credentials=exclude,
|
|
)
|
|
_store_challenge(options.challenge)
|
|
return JSONResponse(content=json.loads(options_to_json(options)))
|
|
|
|
@router.post("/passkey/register/verify")
|
|
async def passkey_register_verify(request: Request, current_user = Depends(auth.get_current_user)):
|
|
body = await request.json()
|
|
challenge = _get_challenge(body.get("response", {}).get("clientDataJSON", ""))
|
|
try:
|
|
verification = verify_registration_response(
|
|
credential=body,
|
|
expected_challenge=challenge,
|
|
expected_rp_id=config.WEBAUTHN_RP_ID,
|
|
expected_origin=config.WEBAUTHN_ORIGINS,
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
auth.save_passkey_credential({
|
|
"credential_id": bytes_to_base64url(verification.credential_id),
|
|
"public_key": bytes_to_base64url(verification.credential_public_key),
|
|
"sign_count": verification.sign_count,
|
|
"name": body.get("name", "Passkey"),
|
|
"created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
})
|
|
return {"ok": True}
|
|
|
|
@router.post("/passkey/login/options")
|
|
@limiter.limit("10/minute")
|
|
async def passkey_login_options(request: Request):
|
|
creds = auth.load_passkey_credentials()
|
|
if not creds:
|
|
raise HTTPException(status_code=404, detail="No passkeys registered")
|
|
allow = [PublicKeyCredentialDescriptor(id=base64url_to_bytes(c["credential_id"])) for c in creds]
|
|
options = generate_authentication_options(
|
|
rp_id=config.WEBAUTHN_RP_ID,
|
|
allow_credentials=allow,
|
|
user_verification=UserVerificationRequirement.PREFERRED,
|
|
)
|
|
_store_challenge(options.challenge)
|
|
return JSONResponse(content=json.loads(options_to_json(options)))
|
|
|
|
@router.post("/passkey/login/verify")
|
|
@limiter.limit("10/minute")
|
|
async def passkey_login_verify(request: Request):
|
|
body = await request.json()
|
|
challenge = _get_challenge(body.get("response", {}).get("clientDataJSON", ""))
|
|
creds = auth.load_passkey_credentials()
|
|
cred_id_b64 = body.get("id", "")
|
|
matched = next((c for c in creds if c["credential_id"] == cred_id_b64), None)
|
|
if not matched:
|
|
raise HTTPException(status_code=400, detail="Unknown credential")
|
|
try:
|
|
verification = verify_authentication_response(
|
|
credential=body,
|
|
expected_challenge=challenge,
|
|
expected_rp_id=config.WEBAUTHN_RP_ID,
|
|
expected_origin=config.WEBAUTHN_ORIGINS,
|
|
credential_public_key=base64url_to_bytes(matched["public_key"]),
|
|
credential_current_sign_count=matched["sign_count"],
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
matched["sign_count"] = verification.new_sign_count
|
|
data = auth._load_auth_data()
|
|
data["passkey_credentials"] = creds
|
|
auth._save_auth_data(data)
|
|
username = config.ADMIN_USER
|
|
access_token = auth.create_access_token(
|
|
data={"sub": username, "role": "admin"},
|
|
expires_delta=timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES),
|
|
)
|
|
refresh_token = auth.create_refresh_token(data={"sub": username, "role": "admin"})
|
|
return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer", "user": username, "role": "admin"}
|
|
|
|
@router.get("/passkey/list")
|
|
async def passkey_list(current_user = Depends(auth.get_current_user)):
|
|
creds = auth.load_passkey_credentials()
|
|
return {"passkeys": [{"id": c["credential_id"], "name": c.get("name", "Passkey"), "created_at": c.get("created_at", "")} for c in creds]}
|
|
|
|
@router.post("/passkey/delete")
|
|
async def passkey_delete(request: Request, current_user = Depends(auth.get_current_user)):
|
|
body = await request.json()
|
|
cred_id = body.get("id")
|
|
data = auth._load_auth_data()
|
|
creds = data.get("passkey_credentials", [])
|
|
data["passkey_credentials"] = [c for c in creds if c["credential_id"] != cred_id]
|
|
auth._save_auth_data(data)
|
|
return {"ok": True}
|
|
|
|
@router.post("/passkey/clear")
|
|
async def passkey_clear(current_user = Depends(auth.get_current_user)):
|
|
auth.clear_passkey_credentials()
|
|
return {"ok": True}
|
|
|
|
# RBAC admin endpoints
|
|
@router.get("/rbac/config")
|
|
async def rbac_config(current_user = Depends(auth.get_current_user)):
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="Admin only")
|
|
from rbac import load_rbac
|
|
return load_rbac()
|
|
|
|
@router.put("/rbac/overrides/{username}")
|
|
async def rbac_set_override(username: str, request: Request, current_user = Depends(auth.get_current_user)):
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="Admin only")
|
|
from rbac import load_rbac, save_rbac
|
|
body = await request.json()
|
|
pages = body.get("pages")
|
|
if pages is None:
|
|
raise HTTPException(status_code=400, detail="Missing pages field")
|
|
rbac = load_rbac()
|
|
existing = rbac.setdefault("user_overrides", {}).get(username, {})
|
|
existing["pages"] = pages
|
|
rbac["user_overrides"][username] = existing
|
|
save_rbac(rbac)
|
|
return {"ok": True}
|
|
|
|
@router.get("/preferences")
|
|
async def get_preferences(current_user = Depends(auth.get_current_user)):
|
|
from rbac import get_sidebar_order
|
|
order = get_sidebar_order(current_user.username)
|
|
return {"sidebar_order": order}
|
|
|
|
@router.put("/preferences")
|
|
async def save_preferences(request: Request, current_user = Depends(auth.get_current_user)):
|
|
import traceback
|
|
from rbac import save_sidebar_order
|
|
try:
|
|
body = await request.json()
|
|
order = body.get("sidebar_order")
|
|
if order is None or not isinstance(order, dict):
|
|
raise HTTPException(status_code=400, detail="Missing sidebar_order dict")
|
|
save_sidebar_order(current_user.username, order)
|
|
return {"ok": True}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
print(f"Error saving preferences: {e}")
|
|
traceback.print_exc()
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@router.delete("/rbac/overrides/{username}")
|
|
async def rbac_delete_override(username: str, current_user = Depends(auth.get_current_user)):
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="Admin only")
|
|
from rbac import load_rbac, save_rbac
|
|
rbac = load_rbac()
|
|
rbac.get("user_overrides", {}).pop(username, None)
|
|
save_rbac(rbac)
|
|
return {"ok": True}
|
|
|
|
@router.put("/rbac/roles/{role}")
|
|
async def rbac_update_role(role: str, request: Request, current_user = Depends(auth.get_current_user)):
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="Admin only")
|
|
from rbac import load_rbac, save_rbac
|
|
body = await request.json()
|
|
pages = body.get("pages")
|
|
if pages is None:
|
|
raise HTTPException(status_code=400, detail="Missing pages field")
|
|
if pages != "*" and not isinstance(pages, list):
|
|
raise HTTPException(status_code=400, detail="pages must be '*' or a list")
|
|
rbac = load_rbac()
|
|
rbac.setdefault("role_defaults", {})[role] = {"pages": pages}
|
|
save_rbac(rbac)
|
|
return {"ok": True}
|