Bernt
623561d30a
- Backend: Docker-container på port 8082 - Redis: Intern container (ingen exposed port) - Nginx-config: nginx-passwordless.conf (väntar på deploy) - Webb: API_BASE uppdaterad till api.quixzoom.com - Fix: datetime timezone-aware i approve-endpoint - Test: End-to-end flöde verifierat
404 lines
13 KiB
Python
404 lines
13 KiB
Python
"""
|
|
Passwordless authentication routes
|
|
Handles cross-device login between app and web
|
|
"""
|
|
|
|
import os
|
|
import time
|
|
import hashlib
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
from fastapi import APIRouter, HTTPException, Header, Request, Depends
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, Field
|
|
import redis
|
|
|
|
router = APIRouter(prefix="/auth/passwordless", tags=["passwordless"])
|
|
|
|
# Redis connection
|
|
redis_client = redis.Redis(
|
|
host=os.getenv('REDIS_HOST', 'localhost'),
|
|
port=int(os.getenv('REDIS_PORT', 6379)),
|
|
db=int(os.getenv('REDIS_DB', 0)),
|
|
decode_responses=True
|
|
)
|
|
|
|
# Constants
|
|
SESSION_TTL_SECONDS = 900 # 15 minutes
|
|
POLL_INTERVAL_MS = 2000
|
|
RATE_LIMIT_PER_IP = 10 # per minute
|
|
RATE_LIMIT_PER_USER = 5 # per minute
|
|
|
|
|
|
# ─── Request/Response Models ────────────────────────────────────────────────
|
|
|
|
class InitiateRequest(BaseModel):
|
|
client_id: str = Field(default="web-dashboard")
|
|
redirect_url: Optional[str] = Field(default="https://quixzoom.se/dashboard")
|
|
device_info: Optional[dict] = Field(default=None)
|
|
|
|
|
|
class InitiateResponse(BaseModel):
|
|
session_id: str
|
|
request_token: str
|
|
qr_data: str
|
|
expires_at: str
|
|
poll_interval_ms: int = POLL_INTERVAL_MS
|
|
|
|
|
|
class ApproveRequest(BaseModel):
|
|
session_id: str
|
|
request_token: str
|
|
signature: str
|
|
timestamp: str
|
|
approving_device_id: str
|
|
|
|
|
|
class ApproveResponse(BaseModel):
|
|
status: str
|
|
session_id: str
|
|
approved_at: str
|
|
|
|
|
|
class StatusResponse(BaseModel):
|
|
status: str
|
|
session_id: str
|
|
request_token: str
|
|
expires_at: str
|
|
approved_at: Optional[str] = None
|
|
completed_at: Optional[str] = None
|
|
access_token: Optional[str] = None
|
|
refresh_token: Optional[str] = None
|
|
error_message: Optional[str] = None
|
|
|
|
|
|
class RejectRequest(BaseModel):
|
|
session_id: str
|
|
request_token: str
|
|
device_id: str
|
|
|
|
|
|
# ─── Helper Functions ───────────────────────────────────────────────────────
|
|
|
|
def generate_session_id() -> str:
|
|
"""Generate unique session ID"""
|
|
return f"pls_{secrets.token_urlsafe(16)}"
|
|
|
|
def generate_request_token() -> str:
|
|
"""Generate cryptographically secure request token"""
|
|
return secrets.token_urlsafe(32)
|
|
|
|
def generate_nonce() -> str:
|
|
"""Generate nonce for challenge"""
|
|
return secrets.token_hex(16)
|
|
|
|
def get_client_ip(request: Request) -> str:
|
|
"""Extract client IP from request"""
|
|
forwarded = request.headers.get('X-Forwarded-For')
|
|
if forwarded:
|
|
return forwarded.split(',')[0].strip()
|
|
return request.client.host if request.client else "unknown"
|
|
|
|
def check_rate_limit(key: str, limit: int, window: int = 60) -> bool:
|
|
"""Check if rate limit is exceeded"""
|
|
current = redis_client.get(f"ratelimit:{key}")
|
|
if not current:
|
|
redis_client.setex(f"ratelimit:{key}", window, 1)
|
|
return True
|
|
|
|
count = int(current)
|
|
if count >= limit:
|
|
return False
|
|
|
|
redis_client.incr(f"ratelimit:{key}")
|
|
return True
|
|
|
|
def create_qr_data(session_id: str, request_token: str) -> str:
|
|
"""Create QR code data string"""
|
|
return f"quixzoom://auth?sid={session_id}&token={request_token}"
|
|
|
|
def hash_fingerprint(user_agent: str, ip: str) -> str:
|
|
"""Create hash of device fingerprint"""
|
|
data = f"{user_agent}:{ip}"
|
|
return hashlib.sha256(data.encode()).hexdigest()
|
|
|
|
|
|
# ─── Routes ─────────────────────────────────────────────────────────────────
|
|
|
|
@router.post("/initiate", response_model=InitiateResponse)
|
|
async def initiate_passwordless(
|
|
request: Request,
|
|
body: InitiateRequest
|
|
):
|
|
"""
|
|
Initiate a new passwordless authentication session.
|
|
Returns session details including QR code data.
|
|
"""
|
|
client_ip = get_client_ip(request)
|
|
user_agent = request.headers.get('User-Agent', 'unknown')
|
|
|
|
# Rate limiting
|
|
if not check_rate_limit(f"ip:{client_ip}", RATE_LIMIT_PER_IP):
|
|
raise HTTPException(
|
|
status_code=429,
|
|
detail="Too many requests. Please try again later."
|
|
)
|
|
|
|
# Generate session
|
|
session_id = generate_session_id()
|
|
request_token = generate_request_token()
|
|
nonce = generate_nonce()
|
|
challenge = secrets.token_urlsafe(32)
|
|
expires_at = datetime.utcnow() + timedelta(seconds=SESSION_TTL_SECONDS)
|
|
|
|
# Store in Redis
|
|
session_data = {
|
|
"session_id": session_id,
|
|
"request_token": request_token,
|
|
"nonce": nonce,
|
|
"challenge": challenge,
|
|
"status": "pending",
|
|
"client_id": body.client_id,
|
|
"redirect_url": body.redirect_url or "https://quixzoom.se/dashboard",
|
|
"web_ip": client_ip,
|
|
"web_user_agent": user_agent,
|
|
"web_fingerprint": hash_fingerprint(user_agent, client_ip),
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
"expires_at": expires_at.isoformat(),
|
|
}
|
|
|
|
redis_key = f"passwordless:{session_id}"
|
|
redis_client.hset(redis_key, mapping=session_data)
|
|
redis_client.expire(redis_key, SESSION_TTL_SECONDS)
|
|
|
|
# Also store by request_token for lookup
|
|
redis_client.setex(
|
|
f"passwordless_token:{request_token}",
|
|
SESSION_TTL_SECONDS,
|
|
session_id
|
|
)
|
|
|
|
return InitiateResponse(
|
|
session_id=session_id,
|
|
request_token=request_token,
|
|
qr_data=create_qr_data(session_id, request_token),
|
|
expires_at=expires_at.isoformat() + "Z",
|
|
poll_interval_ms=POLL_INTERVAL_MS
|
|
)
|
|
|
|
|
|
@router.post("/approve", response_model=ApproveResponse)
|
|
async def approve_passwordless(
|
|
request: Request,
|
|
body: ApproveRequest,
|
|
authorization: Optional[str] = Header(None)
|
|
):
|
|
"""
|
|
Approve a passwordless authentication request from the app.
|
|
Requires valid app JWT in Authorization header.
|
|
"""
|
|
# Verify app JWT (simplified - integrate with your JWT validation)
|
|
if not authorization or not authorization.startswith("Bearer "):
|
|
raise HTTPException(status_code=401, detail="Missing or invalid authorization")
|
|
|
|
app_token = authorization.replace("Bearer ", "")
|
|
|
|
# TODO: Validate app_token against your JWT service
|
|
# For now, we'll do basic validation
|
|
if len(app_token) < 10:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
|
|
# Look up session
|
|
redis_key = f"passwordless:{body.session_id}"
|
|
session_data = redis_client.hgetall(redis_key)
|
|
|
|
if not session_data:
|
|
raise HTTPException(status_code=410, detail="Session not found or expired")
|
|
|
|
# Verify request_token matches
|
|
if session_data.get("request_token") != body.request_token:
|
|
raise HTTPException(status_code=403, detail="Invalid request token")
|
|
|
|
# Check status
|
|
if session_data.get("status") != "pending":
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"Session already {session_data.get('status')}"
|
|
)
|
|
|
|
# Verify timestamp is recent (within 5 minutes)
|
|
try:
|
|
# Parse timestamp and make it offset-aware
|
|
timestamp_str = body.timestamp.replace('Z', '+00:00')
|
|
timestamp = datetime.fromisoformat(timestamp_str)
|
|
|
|
# Ensure timestamp is offset-aware
|
|
if timestamp.tzinfo is None:
|
|
timestamp = timestamp.replace(tzinfo=timezone.utc)
|
|
|
|
# Get current UTC time as offset-aware
|
|
now = datetime.now(timezone.utc)
|
|
|
|
if now - timestamp > timedelta(minutes=5):
|
|
raise HTTPException(status_code=403, detail="Timestamp too old")
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=f"Invalid timestamp format: {str(e)}")
|
|
|
|
# TODO: Verify signature
|
|
# expected_signature = hmac_sha256(session_id + request_token + timestamp, device_secret)
|
|
# For MVP, we'll skip signature verification and rely on JWT + request_token
|
|
|
|
# Update session
|
|
approved_at = datetime.utcnow()
|
|
redis_client.hset(redis_key, mapping={
|
|
"status": "approved",
|
|
"approved_at": approved_at.isoformat(),
|
|
"approving_device_id": body.approving_device_id,
|
|
"device_ip": get_client_ip(request),
|
|
"device_user_agent": request.headers.get('User-Agent', 'unknown'),
|
|
})
|
|
|
|
# Keep TTL but extend slightly for completion
|
|
redis_client.expire(redis_key, 300) # 5 more minutes
|
|
|
|
return ApproveResponse(
|
|
status="approved",
|
|
session_id=body.session_id,
|
|
approved_at=approved_at.isoformat() + "Z"
|
|
)
|
|
|
|
|
|
@router.get("/status")
|
|
async def check_status(
|
|
request: Request,
|
|
session_id: str,
|
|
request_token: str
|
|
):
|
|
"""
|
|
Check the status of a passwordless authentication session.
|
|
Called by web client via polling.
|
|
"""
|
|
redis_key = f"passwordless:{session_id}"
|
|
session_data = redis_client.hgetall(redis_key)
|
|
|
|
if not session_data:
|
|
raise HTTPException(status_code=404, detail="Session not found")
|
|
|
|
# Verify request_token
|
|
if session_data.get("request_token") != request_token:
|
|
raise HTTPException(status_code=403, detail="Invalid request token")
|
|
|
|
status = session_data.get("status", "unknown")
|
|
|
|
response = {
|
|
"status": status,
|
|
"session_id": session_id,
|
|
"request_token": request_token,
|
|
"expires_at": session_data.get("expires_at"),
|
|
}
|
|
|
|
if status == "approved":
|
|
response["approved_at"] = session_data.get("approved_at")
|
|
|
|
# Generate web session tokens
|
|
# TODO: Integrate with your JWT service to create proper tokens
|
|
user_id = "user_from_app_token" # Extract from app JWT
|
|
|
|
# Create web session
|
|
web_session_token = secrets.token_urlsafe(32)
|
|
refresh_token = secrets.token_urlsafe(32)
|
|
|
|
# Store web session
|
|
web_session_key = f"web_session:{web_session_token}"
|
|
redis_client.hset(web_session_key, mapping={
|
|
"user_id": user_id,
|
|
"session_id": session_id,
|
|
"created_at": datetime.utcnow().isoformat(),
|
|
"ip": get_client_ip(request),
|
|
"user_agent": request.headers.get('User-Agent', 'unknown'),
|
|
})
|
|
redis_client.expire(web_session_key, 86400) # 24 hours
|
|
|
|
# Mark session as completed
|
|
redis_client.hset(redis_key, mapping={
|
|
"status": "completed",
|
|
"completed_at": datetime.utcnow().isoformat(),
|
|
"web_session_token": web_session_token,
|
|
})
|
|
|
|
response.update({
|
|
"status": "completed",
|
|
"completed_at": datetime.utcnow().isoformat() + "Z",
|
|
"access_token": web_session_token,
|
|
"refresh_token": refresh_token,
|
|
})
|
|
|
|
elif status == "rejected":
|
|
response["error_message"] = "Authentication rejected by user"
|
|
|
|
elif status == "expired":
|
|
response["error_message"] = "Session expired"
|
|
|
|
return StatusResponse(**response)
|
|
|
|
|
|
@router.post("/reject")
|
|
async def reject_passwordless(
|
|
request: Request,
|
|
body: RejectRequest,
|
|
authorization: Optional[str] = Header(None)
|
|
):
|
|
"""
|
|
Reject a passwordless authentication request from the app.
|
|
"""
|
|
# Verify app JWT
|
|
if not authorization or not authorization.startswith("Bearer "):
|
|
raise HTTPException(status_code=401, detail="Missing authorization")
|
|
|
|
redis_key = f"passwordless:{body.session_id}"
|
|
session_data = redis_client.hgetall(redis_key)
|
|
|
|
if not session_data:
|
|
raise HTTPException(status_code=404, detail="Session not found")
|
|
|
|
if session_data.get("request_token") != body.request_token:
|
|
raise HTTPException(status_code=403, detail="Invalid request token")
|
|
|
|
redis_client.hset(redis_key, mapping={
|
|
"status": "rejected",
|
|
"rejected_at": datetime.utcnow().isoformat(),
|
|
"rejecting_device_id": body.device_id,
|
|
})
|
|
|
|
return {"status": "rejected", "session_id": body.session_id}
|
|
|
|
|
|
@router.post("/cancel")
|
|
async def cancel_passwordless(
|
|
session_id: str,
|
|
request_token: str
|
|
):
|
|
"""
|
|
Cancel a pending passwordless authentication request.
|
|
Called by web client.
|
|
"""
|
|
redis_key = f"passwordless:{session_id}"
|
|
session_data = redis_client.hgetall(redis_key)
|
|
|
|
if not session_data:
|
|
raise HTTPException(status_code=404, detail="Session not found")
|
|
|
|
if session_data.get("request_token") != request_token:
|
|
raise HTTPException(status_code=403, detail="Invalid request token")
|
|
|
|
if session_data.get("status") != "pending":
|
|
raise HTTPException(status_code=409, detail="Session already processed")
|
|
|
|
redis_client.hset(redis_key, mapping={
|
|
"status": "cancelled",
|
|
"cancelled_at": datetime.utcnow().isoformat(),
|
|
})
|
|
|
|
return {"status": "cancelled", "session_id": session_id}
|