chore: initial project setup with backend, frontend, Android app, and CI/CD
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
This commit is contained in:
@@ -0,0 +1,14 @@
|
||||
import os
|
||||
|
||||
os.environ["DATABASE_URL"] = "sqlite+aiosqlite:///:memory:"
|
||||
os.environ["ENVIRONMENT"] = "test"
|
||||
|
||||
from app.main import app
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
def test_health_endpoint():
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/health")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["status"] == "ok"
|
||||
@@ -0,0 +1,80 @@
|
||||
from datetime import timedelta
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core.security import now_utc
|
||||
from app.models.challenge import Challenge, ChallengeStatus
|
||||
from app.services.challenges import approve_challenge, expire_if_needed
|
||||
|
||||
|
||||
class DummySession:
|
||||
def __init__(self, device=None):
|
||||
self.device = device
|
||||
self.added = []
|
||||
|
||||
def add(self, item):
|
||||
self.added.append(item)
|
||||
|
||||
async def get(self, model, ident):
|
||||
return self.device
|
||||
|
||||
|
||||
class DummyDevice:
|
||||
def __init__(self, device_id, user_id):
|
||||
self.id = device_id
|
||||
self.user_id = user_id
|
||||
self.is_revoked = False
|
||||
self.public_key_pem = "invalid"
|
||||
self.last_seen_at = None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_expired_challenge_transitions_to_expired():
|
||||
challenge = Challenge(
|
||||
id=uuid4(),
|
||||
user_id=uuid4(),
|
||||
device_id=uuid4(),
|
||||
status=ChallengeStatus.pending,
|
||||
relying_party="app",
|
||||
requester_ip="127.0.0.1",
|
||||
nonce="nonce",
|
||||
payload={},
|
||||
issued_at=now_utc() - timedelta(seconds=120),
|
||||
expires_at=now_utc() - timedelta(seconds=1),
|
||||
)
|
||||
|
||||
changed = await expire_if_needed(DummySession(), challenge)
|
||||
|
||||
assert changed
|
||||
assert challenge.status == ChallengeStatus.expired
|
||||
assert challenge.responded_at is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_replay_is_blocked_after_approval_state():
|
||||
device_id = uuid4()
|
||||
challenge = Challenge(
|
||||
id=uuid4(),
|
||||
user_id=uuid4(),
|
||||
device_id=device_id,
|
||||
status=ChallengeStatus.approved,
|
||||
relying_party="app",
|
||||
requester_ip="127.0.0.1",
|
||||
nonce="nonce",
|
||||
payload={"challenge_id": "x"},
|
||||
issued_at=now_utc(),
|
||||
expires_at=now_utc() + timedelta(seconds=60),
|
||||
)
|
||||
|
||||
ok, reason = await approve_challenge(
|
||||
DummySession(DummyDevice(device_id, challenge.user_id)),
|
||||
challenge=challenge,
|
||||
device_id=device_id,
|
||||
payload=challenge.payload,
|
||||
signature="anything",
|
||||
ip_address="127.0.0.1",
|
||||
)
|
||||
|
||||
assert not ok
|
||||
assert reason == "challenge_not_pending"
|
||||
@@ -0,0 +1,22 @@
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature
|
||||
from cryptography.hazmat.primitives.hashes import SHA256
|
||||
|
||||
from app.core.security import b64url, canonical_json, verify_signature
|
||||
|
||||
|
||||
def test_verify_es256_signature_der_and_raw():
|
||||
private_key = ec.generate_private_key(ec.SECP256R1())
|
||||
public_pem = private_key.public_key().public_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PublicFormat.SubjectPublicKeyInfo,
|
||||
).decode("utf-8")
|
||||
payload = {"challenge_id": "abc", "nonce": "n", "user_id": "u"}
|
||||
der_sig = private_key.sign(canonical_json(payload), ec.ECDSA(SHA256()))
|
||||
r, s = decode_dss_signature(der_sig)
|
||||
raw_sig = r.to_bytes(32, "big") + s.to_bytes(32, "big")
|
||||
|
||||
assert verify_signature(public_pem, payload, b64url(der_sig))
|
||||
assert verify_signature(public_pem, payload, b64url(raw_sig))
|
||||
assert not verify_signature(public_pem, payload | {"nonce": "changed"}, b64url(der_sig))
|
||||
Reference in New Issue
Block a user