nessi
f925009977
Add complete NexaMFA push MFA system with: - FastAPI backend with PostgreSQL, Redis, OIDC provider, and Prometheus metrics - React TypeScript admin console - Android Kotlin/Jetpack Compose app with biometric authentication - Docker Compose deployment configuration - Gitea CI workflow for backend, frontend, and Android builds - Environment configuration template with security settings - Documentation for security model, deployment
95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
import base64
|
|
import hashlib
|
|
import json
|
|
from io import BytesIO
|
|
|
|
import qrcode
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import Settings
|
|
from app.core.security import expires_in, random_token, now_utc
|
|
from app.models.device import Device
|
|
from app.models.enrollment import Enrollment
|
|
from app.models.user import User
|
|
from app.services.audit import audit
|
|
|
|
|
|
def token_hash(token: str) -> str:
|
|
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
|
|
|
|
|
async def start_enrollment(
|
|
session: AsyncSession,
|
|
settings: Settings,
|
|
*,
|
|
username: str,
|
|
display_name: str | None,
|
|
email: str | None,
|
|
ip_address: str | None,
|
|
) -> tuple[Enrollment, dict, str]:
|
|
user = await session.scalar(select(User).where(User.username == username))
|
|
if not user:
|
|
user = User(username=username, display_name=display_name, email=email)
|
|
session.add(user)
|
|
await session.flush()
|
|
|
|
raw_token = random_token(32)
|
|
enrollment = Enrollment(
|
|
user_id=user.id,
|
|
token_hash=token_hash(raw_token),
|
|
expires_at=expires_in(settings.enrollment_ttl_seconds),
|
|
)
|
|
session.add(enrollment)
|
|
await session.flush()
|
|
|
|
qr_payload = {
|
|
"type": "nexamfa-enrollment",
|
|
"server_url": str(settings.public_base_url).rstrip("/"),
|
|
"enrollment_id": str(enrollment.id),
|
|
"enrollment_token": raw_token,
|
|
"username": user.username,
|
|
"expires_at": enrollment.expires_at.isoformat(),
|
|
}
|
|
qr = qrcode.make(json.dumps(qr_payload, separators=(",", ":")))
|
|
buf = BytesIO()
|
|
qr.save(buf, format="PNG")
|
|
qr_b64 = base64.b64encode(buf.getvalue()).decode("ascii")
|
|
await audit(session, "enrollment.started", actor=username, target_type="user", target_id=str(user.id), ip_address=ip_address)
|
|
return enrollment, qr_payload, qr_b64
|
|
|
|
|
|
async def finish_enrollment(
|
|
session: AsyncSession,
|
|
*,
|
|
enrollment_token: str,
|
|
device_name: str,
|
|
public_key_pem: str,
|
|
public_key_alg: str,
|
|
fcm_token: str | None,
|
|
app_version: str | None,
|
|
attestation: dict | None,
|
|
ip_address: str | None,
|
|
) -> Device | None:
|
|
enrollment = await session.scalar(
|
|
select(Enrollment).where(Enrollment.token_hash == token_hash(enrollment_token))
|
|
)
|
|
if not enrollment or enrollment.used or enrollment.expires_at <= now_utc():
|
|
return None
|
|
|
|
device = Device(
|
|
user_id=enrollment.user_id,
|
|
name=device_name,
|
|
public_key_pem=public_key_pem,
|
|
public_key_alg=public_key_alg,
|
|
fcm_token=fcm_token,
|
|
app_version=app_version,
|
|
attestation=attestation,
|
|
last_seen_at=now_utc(),
|
|
)
|
|
enrollment.used = True
|
|
session.add(device)
|
|
await session.flush()
|
|
await audit(session, "device.enrolled", actor=str(enrollment.user_id), target_type="device", target_id=str(device.id), ip_address=ip_address)
|
|
return device
|