Phase 7: Auth & Security upgrade (JWT, 2FA, Login UI)
This commit is contained in:
@@ -0,0 +1,86 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from pydantic import BaseModel
|
||||
import config
|
||||
import auth
|
||||
import pyotp
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
class LoginRequest(BaseModel):
|
||||
username: str
|
||||
password: str
|
||||
totp_code: str = ""
|
||||
|
||||
class Token(BaseModel):
|
||||
access_token: str
|
||||
token_type: str
|
||||
user: str
|
||||
has_2fa: bool
|
||||
|
||||
class Setup2FAResponse(BaseModel):
|
||||
secret: str
|
||||
uri: str
|
||||
|
||||
class Verify2FARequest(BaseModel):
|
||||
secret: str
|
||||
code: str
|
||||
|
||||
@router.post("/login", response_model=Token)
|
||||
async def login(creds: LoginRequest):
|
||||
# Verify username/password
|
||||
if creds.username != config.ADMIN_USER or not auth.verify_password(creds.password, config.ADMIN_PASSWORD_HASH):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Incorrect username or password",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
# Check 2FA
|
||||
totp_secret = auth.load_totp_secret()
|
||||
if totp_secret:
|
||||
if not creds.totp_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):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid 2FA code",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
access_token = auth.create_access_token(data={"sub": creds.username})
|
||||
return {
|
||||
"access_token": access_token,
|
||||
"token_type": "bearer",
|
||||
"user": creds.username,
|
||||
"has_2fa": bool(totp_secret)
|
||||
}
|
||||
|
||||
@router.get("/me")
|
||||
async def read_users_me(current_user: str = Depends(auth.get_current_user)):
|
||||
totp_secret = auth.load_totp_secret()
|
||||
return {"username": current_user, "has_2fa": bool(totp_secret)}
|
||||
|
||||
@router.post("/setup-2fa/generate", response_model=Setup2FAResponse)
|
||||
async def generate_2fa(current_user: str = Depends(auth.get_current_user)):
|
||||
secret = pyotp.random_base32()
|
||||
uri = pyotp.totp.TOTP(secret).provisioning_uri(name=current_user, issuer_name="NAS Dashboard")
|
||||
return {"secret": secret, "uri": uri}
|
||||
|
||||
@router.post("/setup-2fa/verify")
|
||||
async def verify_2fa_setup(request: Verify2FARequest, current_user: str = 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: str = Depends(auth.get_current_user)):
|
||||
auth.save_totp_secret("")
|
||||
return {"message": "2FA disabled"}
|
||||
Reference in New Issue
Block a user