WebAuthn

// WebAuthn 서버 구현 - Node.js (SimpleWebAuthn 라이브러리)
import {
    generateRegistrationOptions,
    verifyRegistrationResponse,
    generateAuthenticationOptions,
    verifyAuthenticationResponse
} from '@simplewebauthn/server';

const rpName = 'KAITRUST';
const rpID = 'kaitrust.ai';
const origin = 'https://kaitrust.ai';

// 등록 옵션 생성
app.post('/api/webauthn/register/options', async (req, res) => {
    const { username } = req.body;
    const user = await db.findOrCreateUser(username);

    const options = await generateRegistrationOptions({
        rpName,
        rpID,
        userID: user.id,
        userName: username,
        userDisplayName: username,
        attestationType: 'none',
        authenticatorSelection: {
            authenticatorAttachment: 'platform',
            userVerification: 'required',
            residentKey: 'required'
        },
        supportedAlgorithmIDs: [-7, -257]  // ES256, RS256
    });

    // 챌린지 임시 저장 (세션 또는 Redis)
    await redis.set(`webauthn:challenge:${user.id}`, options.challenge, 'EX', 300);

    res.json(options);
});

// 등록 검증
app.post('/api/webauthn/register/verify', async (req, res) => {
    const { body } = req;
    const user = req.session.user;
    const expectedChallenge = await redis.get(`webauthn:challenge:${user.id}`);

    try {
        const verification = await verifyRegistrationResponse({
            response: body,
            expectedChallenge,
            expectedOrigin: origin,
            expectedRPID: rpID
        });

        if (verification.verified) {
            // 공개키 저장
            await db.saveCredential({
                credentialID: verification.registrationInfo.credentialID,
                credentialPublicKey: verification.registrationInfo.credentialPublicKey,
                counter: verification.registrationInfo.counter,
                userId: user.id
            });

            res.json({ success: true });
        }
    } catch (error) {
        res.status(400).json({ error: error.message });
    }
});

// 인증 옵션 생성
app.get('/api/webauthn/authenticate/options', async (req, res) => {
    const options = await generateAuthenticationOptions({
        rpID,
        userVerification: 'required',
        timeout: 60000
    });

    await redis.set(`webauthn:auth:challenge`, options.challenge, 'EX', 300);
    res.json(options);
});

// 인증 검증
app.post('/api/webauthn/authenticate/verify', async (req, res) => {
    const { body } = req;
    const expectedChallenge = await redis.get(`webauthn:auth:challenge`);
    const credential = await db.findCredential(body.id);

    const verification = await verifyAuthenticationResponse({
        response: body,
        expectedChallenge,
        expectedOrigin: origin,
        expectedRPID: rpID,
        authenticator: {
            credentialID: credential.credentialID,
            credentialPublicKey: credential.credentialPublicKey,
            counter: credential.counter
        }
    });

    if (verification.verified) {
        // 카운터 업데이트 (리플레이 공격 방지)
        await db.updateCredentialCounter(
            credential.id,
            verification.authenticationInfo.newCounter
        );

        // JWT 토큰 발급
        const token = jwt.sign({ userId: credential.userId }, process.env.JWT_SECRET);
        res.json({ success: true, token });
    }
});

# WebAuthn 서버 검증 - Python (py_webauthn 라이브러리)
from webauthn import (
    generate_registration_options,
    verify_registration_response,
    generate_authentication_options,
    verify_authentication_response,
    options_to_json
)
from webauthn.helpers.structs import (
    AuthenticatorSelectionCriteria,
    UserVerificationRequirement,
    ResidentKeyRequirement,
    PublicKeyCredentialDescriptor
)
import json

RP_ID = "kaitrust.ai"
RP_NAME = "KAITRUST"
ORIGIN = "https://kaitrust.ai"

# 등록 옵션 생성
def create_registration_options(user_id: str, username: str):
    options = generate_registration_options(
        rp_id=RP_ID,
        rp_name=RP_NAME,
        user_id=user_id.encode(),
        user_name=username,
        user_display_name=username,
        authenticator_selection=AuthenticatorSelectionCriteria(
            user_verification=UserVerificationRequirement.REQUIRED,
            resident_key=ResidentKeyRequirement.REQUIRED
        )
    )
    # 챌린지 저장 (Redis 또는 세션)
    redis_client.setex(
        f"webauthn:register:{user_id}",
        300,
        options.challenge
    )
    return options_to_json(options)

# 등록 응답 검증
def verify_registration(user_id: str, credential: dict):
    expected_challenge = redis_client.get(f"webauthn:register:{user_id}")

    verification = verify_registration_response(
        credential=credential,
        expected_challenge=expected_challenge,
        expected_rp_id=RP_ID,
        expected_origin=ORIGIN,
        require_user_verification=True
    )

    # DB에 크레덴셜 저장
    db.credentials.insert_one({
        "user_id": user_id,
        "credential_id": verification.credential_id,
        "public_key": verification.credential_public_key,
        "sign_count": verification.sign_count,
        "created_at": datetime.utcnow()
    })

    return {"success": True, "credential_id": verification.credential_id.hex()}

# 인증 옵션 생성
def create_authentication_options(user_id: str = None):
    # 사용자의 등록된 크레덴셜 조회
    allow_credentials = []
    if user_id:
        creds = db.credentials.find({"user_id": user_id})
        allow_credentials = [
            PublicKeyCredentialDescriptor(id=c["credential_id"])
            for c in creds
        ]

    options = generate_authentication_options(
        rp_id=RP_ID,
        user_verification=UserVerificationRequirement.REQUIRED,
        allow_credentials=allow_credentials
    )

    redis_client.setex("webauthn:auth:challenge", 300, options.challenge)
    return options_to_json(options)

# 인증 응답 검증
def verify_authentication(credential: dict):
    expected_challenge = redis_client.get("webauthn:auth:challenge")
    stored_cred = db.credentials.find_one({
        "credential_id": bytes.fromhex(credential["id"])
    })

    verification = verify_authentication_response(
        credential=credential,
        expected_challenge=expected_challenge,
        expected_rp_id=RP_ID,
        expected_origin=ORIGIN,
        credential_public_key=stored_cred["public_key"],
        credential_current_sign_count=stored_cred["sign_count"],
        require_user_verification=True
    )

    # 서명 카운터 업데이트
    db.credentials.update_one(
        {"_id": stored_cred["_id"]},
        {"$set": {"sign_count": verification.new_sign_count}}
    )

    return {
        "success": True,
        "user_id": stored_cred["user_id"]
    }