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
139 lines
5.0 KiB
Python
139 lines
5.0 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import Settings
|
|
from app.core.security import expires_in, now_utc, random_token, verify_signature
|
|
from app.models.challenge import Challenge, ChallengeStatus
|
|
from app.models.device import Device
|
|
from app.models.user import User
|
|
from app.services.audit import audit
|
|
from app.services.push import PushService
|
|
|
|
|
|
def build_challenge_payload(challenge: Challenge, username: str) -> dict:
|
|
return {
|
|
"challenge_id": str(challenge.id),
|
|
"user_id": str(challenge.user_id),
|
|
"username": username,
|
|
"relying_party": challenge.relying_party,
|
|
"requester_ip": challenge.requester_ip,
|
|
"location": challenge.location,
|
|
"issued_at": challenge.issued_at.isoformat(),
|
|
"expires_at": challenge.expires_at.isoformat(),
|
|
"nonce": challenge.nonce,
|
|
}
|
|
|
|
|
|
async def create_challenge(
|
|
session: AsyncSession,
|
|
settings: Settings,
|
|
push: PushService,
|
|
*,
|
|
username: str,
|
|
relying_party: str,
|
|
requester_ip: str,
|
|
location: str | None,
|
|
ttl_seconds: int | None,
|
|
oidc_state: str | None = None,
|
|
) -> Challenge | None:
|
|
user = await session.scalar(select(User).where(User.username == username))
|
|
if not user:
|
|
return None
|
|
device = await session.scalar(
|
|
select(Device)
|
|
.where(Device.user_id == user.id, Device.is_revoked.is_(False))
|
|
.order_by(Device.last_seen_at.desc().nullslast(), Device.created_at.desc())
|
|
)
|
|
if not device:
|
|
return None
|
|
|
|
issued = now_utc()
|
|
challenge = Challenge(
|
|
user_id=user.id,
|
|
device_id=device.id,
|
|
relying_party=relying_party,
|
|
requester_ip=requester_ip,
|
|
location=location,
|
|
nonce=random_token(32),
|
|
issued_at=issued,
|
|
expires_at=expires_in(ttl_seconds or settings.challenge_ttl_seconds),
|
|
oidc_state=oidc_state,
|
|
payload={},
|
|
)
|
|
session.add(challenge)
|
|
await session.flush()
|
|
challenge.payload = build_challenge_payload(challenge, user.username)
|
|
await audit(
|
|
session,
|
|
"challenge.created",
|
|
actor=user.username,
|
|
target_type="challenge",
|
|
target_id=str(challenge.id),
|
|
ip_address=requester_ip,
|
|
metadata={"device_id": str(device.id), "oidc": bool(oidc_state)},
|
|
)
|
|
await push.send_challenge(device.fcm_token, str(challenge.id))
|
|
return challenge
|
|
|
|
|
|
async def expire_if_needed(session: AsyncSession, challenge: Challenge) -> bool:
|
|
if challenge.status == ChallengeStatus.pending and challenge.expires_at <= now_utc():
|
|
challenge.status = ChallengeStatus.expired
|
|
challenge.responded_at = now_utc()
|
|
await audit(session, "challenge.expired", target_type="challenge", target_id=str(challenge.id))
|
|
return True
|
|
return False
|
|
|
|
|
|
async def approve_challenge(
|
|
session: AsyncSession,
|
|
*,
|
|
challenge: Challenge,
|
|
device_id,
|
|
payload: dict,
|
|
signature: str,
|
|
ip_address: str | None,
|
|
) -> tuple[bool, str]:
|
|
await expire_if_needed(session, challenge)
|
|
if challenge.status != ChallengeStatus.pending:
|
|
await audit(session, "challenge.replay_blocked", target_type="challenge", target_id=str(challenge.id), ip_address=ip_address)
|
|
return False, "challenge_not_pending"
|
|
if str(device_id) != str(challenge.device_id):
|
|
return False, "wrong_device"
|
|
|
|
device = await session.get(Device, device_id)
|
|
if not device or device.is_revoked:
|
|
await audit(session, "challenge.revoked_device_blocked", target_type="challenge", target_id=str(challenge.id), ip_address=ip_address)
|
|
return False, "device_revoked"
|
|
if payload != challenge.payload:
|
|
return False, "payload_mismatch"
|
|
if not verify_signature(device.public_key_pem, payload, signature):
|
|
await audit(session, "challenge.signature_invalid", target_type="challenge", target_id=str(challenge.id), ip_address=ip_address)
|
|
return False, "invalid_signature"
|
|
|
|
challenge.status = ChallengeStatus.approved
|
|
challenge.signature = signature
|
|
challenge.responded_at = now_utc()
|
|
device.last_seen_at = now_utc()
|
|
await audit(session, "challenge.approved", actor=str(device.user_id), target_type="challenge", target_id=str(challenge.id), ip_address=ip_address, metadata={"device_id": str(device.id)})
|
|
return True, "approved"
|
|
|
|
|
|
async def deny_challenge(
|
|
session: AsyncSession,
|
|
*,
|
|
challenge: Challenge,
|
|
device_id,
|
|
reason: str | None,
|
|
ip_address: str | None,
|
|
) -> tuple[bool, str]:
|
|
await expire_if_needed(session, challenge)
|
|
if challenge.status != ChallengeStatus.pending:
|
|
return False, "challenge_not_pending"
|
|
if device_id and str(device_id) != str(challenge.device_id):
|
|
return False, "wrong_device"
|
|
challenge.status = ChallengeStatus.denied
|
|
challenge.responded_at = now_utc()
|
|
await audit(session, "challenge.denied", target_type="challenge", target_id=str(challenge.id), ip_address=ip_address, metadata={"reason": reason})
|
|
return True, "denied"
|