Cheng Zhou
660 lines
23 KiB
Python
660 lines
23 KiB
Python
import pytest
|
|
import pytest_asyncio
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
|
from fastapi import HTTPException
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import UUID as SA_UUID, select
|
|
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
from datetime import datetime, timedelta, timezone
|
|
import base64
|
|
import json
|
|
import os
|
|
|
|
from app.main import create_app
|
|
from app.core.config import settings
|
|
from app.core.deps import get_db_session
|
|
from app.core.security import hash_password, verify_password
|
|
from app.crud import user as user_crud
|
|
from app.db.base_class import Base
|
|
from app.models.audit_log import AuditLog
|
|
from app.models.email_settings import (
|
|
EmailVerificationCode,
|
|
EmailVerificationPurpose,
|
|
SmtpSecurity,
|
|
SystemEmailSettings,
|
|
)
|
|
from app.services import email_service
|
|
from tests.conftest import GUID
|
|
from app.models.study_member import StudyMember
|
|
from app.models.user import User, UserStatus
|
|
from app.schemas.user import UserRegisterRequest
|
|
|
|
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
|
|
|
|
|
|
async def mark_register_email_verified(SessionLocal, email: str) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
async with SessionLocal() as session:
|
|
session.add(
|
|
EmailVerificationCode(
|
|
email=email,
|
|
purpose=EmailVerificationPurpose.REGISTER,
|
|
code_hash=hash_password("123456"),
|
|
expires_at=now + timedelta(minutes=10),
|
|
verified_at=now,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
async def add_email_settings(SessionLocal, domain: str) -> None:
|
|
async with SessionLocal() as session:
|
|
session.add(
|
|
SystemEmailSettings(
|
|
register_domain=domain,
|
|
smtp_host="smtp.test",
|
|
smtp_port=465,
|
|
smtp_security=SmtpSecurity.SSL,
|
|
smtp_username="mailer",
|
|
smtp_password_encrypted="encrypted",
|
|
sender_email=f"mailer@{domain}",
|
|
allowed_register_domain=domain,
|
|
verification_code_ttl_minutes=10,
|
|
send_cooldown_seconds=60,
|
|
max_verify_attempts=5,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
async def encrypted_auth_payload(client: AsyncClient, email: str, password: str) -> dict:
|
|
key_resp = await client.get("/api/v1/auth/login-key")
|
|
assert key_resp.status_code == 200
|
|
login_key = key_resp.json()
|
|
public_key = serialization.load_pem_public_key(login_key["public_key"].encode("utf-8"))
|
|
plaintext = json.dumps(
|
|
{
|
|
"email": email,
|
|
"password": password,
|
|
"challenge": login_key["challenge"],
|
|
},
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
aes_key = AESGCM.generate_key(bit_length=256)
|
|
iv = os.urandom(12)
|
|
encrypted_data = AESGCM(aes_key).encrypt(iv, plaintext, None)
|
|
encrypted_key = public_key.encrypt(
|
|
aes_key,
|
|
padding.OAEP(
|
|
mgf=padding.MGF1(algorithm=hashes.SHA256()),
|
|
algorithm=hashes.SHA256(),
|
|
label=None,
|
|
),
|
|
)
|
|
return {
|
|
"key_id": login_key["key_id"],
|
|
"challenge": login_key["challenge"],
|
|
"ciphertext": base64.b64encode(
|
|
json.dumps(
|
|
{
|
|
"encrypted_key": base64.b64encode(encrypted_key).decode("ascii"),
|
|
"iv": base64.b64encode(iv).decode("ascii"),
|
|
"data": base64.b64encode(encrypted_data).decode("ascii"),
|
|
},
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
).decode("ascii"),
|
|
}
|
|
|
|
|
|
async def encrypted_login(client: AsyncClient, email: str, password: str):
|
|
return await client.post("/api/v1/auth/login", json=await encrypted_auth_payload(client, email, password))
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client_and_db():
|
|
engine = create_async_engine(TEST_DATABASE_URL, future=True)
|
|
TestSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
|
original_column_types = {
|
|
column: column.type
|
|
for table in Base.metadata.tables.values()
|
|
for column in table.columns
|
|
}
|
|
for table in Base.metadata.tables.values():
|
|
for column in table.columns:
|
|
if isinstance(column.type, (PG_UUID, SA_UUID)):
|
|
column.type = GUID()
|
|
try:
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
async def override_get_db():
|
|
async with TestSessionLocal() as session:
|
|
yield session
|
|
|
|
app = create_app()
|
|
app.dependency_overrides[get_db_session] = override_get_db
|
|
|
|
async with TestSessionLocal() as session:
|
|
admin = User(
|
|
email="admin@test.com",
|
|
password_hash=hash_password("admin123"),
|
|
full_name="Admin",
|
|
clinical_department="Admin",
|
|
is_admin=True,
|
|
status=UserStatus.ACTIVE,
|
|
)
|
|
session.add(admin)
|
|
await session.commit()
|
|
|
|
async with AsyncClient(app=app, base_url="http://test") as client:
|
|
yield client, TestSessionLocal
|
|
finally:
|
|
for column, original_type in original_column_types.items():
|
|
column.type = original_type
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_creates_active_user(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
payload = {
|
|
"email": "newuser@test.com",
|
|
"password": "Password123",
|
|
"full_name": "New User",
|
|
"clinical_department": "Clinical",
|
|
}
|
|
await mark_register_email_verified(SessionLocal, payload["email"])
|
|
resp = await client.post("/api/v1/auth/register", json=payload)
|
|
assert resp.status_code == 201
|
|
async with SessionLocal() as session:
|
|
user = await user_crud.get_by_email(session, payload["email"])
|
|
assert user is not None
|
|
assert user.status == UserStatus.ACTIVE
|
|
assert user.is_admin is False
|
|
member_rows = (
|
|
await session.execute(select(StudyMember).where(StudyMember.user_id == user.id))
|
|
).scalars().all()
|
|
assert member_rows == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_rejects_duplicate_email(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
payload = {
|
|
"email": "duplicate@test.com",
|
|
"password": "Password123",
|
|
"full_name": "Duplicate User",
|
|
"clinical_department": "Clinical",
|
|
}
|
|
await mark_register_email_verified(SessionLocal, payload["email"])
|
|
first_resp = await client.post("/api/v1/auth/register", json=payload)
|
|
assert first_resp.status_code == 201
|
|
|
|
second_resp = await client.post("/api/v1/auth/register", json=payload)
|
|
assert second_resp.status_code == 409
|
|
assert second_resp.json()["detail"] == "邮箱已注册"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_requires_verified_email_code(client_and_db):
|
|
client, _ = client_and_db
|
|
payload = {
|
|
"email": "unverified@test.com",
|
|
"password": "Password123",
|
|
"full_name": "Unverified User",
|
|
"clinical_department": "Clinical",
|
|
}
|
|
resp = await client.post("/api/v1/auth/register", json=payload)
|
|
assert resp.status_code == 400
|
|
assert resp.json()["detail"] == "请先完成邮箱验证码校验"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_email_availability(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
available_resp = await client.get(
|
|
"/api/v1/auth/register/email-availability",
|
|
params={"email": "available@test.com"},
|
|
)
|
|
assert available_resp.status_code == 200
|
|
assert available_resp.json() == {"available": True}
|
|
|
|
payload = {
|
|
"email": "used@test.com",
|
|
"password": "Password123",
|
|
"full_name": "Used User",
|
|
"clinical_department": "Clinical",
|
|
}
|
|
await mark_register_email_verified(SessionLocal, payload["email"])
|
|
await client.post("/api/v1/auth/register", json=payload)
|
|
used_resp = await client.get(
|
|
"/api/v1/auth/register/email-availability",
|
|
params={"email": payload["email"]},
|
|
)
|
|
assert used_resp.status_code == 200
|
|
assert used_resp.json() == {"available": False}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_password_reset_code_is_one_time_and_ignores_link_records(client_and_db):
|
|
_, SessionLocal = client_and_db
|
|
email = "reset-once@test.com"
|
|
now = datetime.now(timezone.utc)
|
|
await add_email_settings(SessionLocal, "test.com")
|
|
async with SessionLocal() as session:
|
|
user = User(
|
|
email=email,
|
|
password_hash=hash_password("OldPassword123"),
|
|
full_name="Reset Once",
|
|
clinical_department="Clinical",
|
|
status=UserStatus.ACTIVE,
|
|
)
|
|
session.add(user)
|
|
session.add_all(
|
|
[
|
|
EmailVerificationCode(
|
|
email=email,
|
|
purpose=EmailVerificationPurpose.PASSWORD_RESET,
|
|
code_hash=hash_password("123456"),
|
|
expires_at=now + timedelta(minutes=10),
|
|
created_at=now,
|
|
),
|
|
EmailVerificationCode(
|
|
email=email,
|
|
purpose=EmailVerificationPurpose.PASSWORD_RESET_LINK,
|
|
code_hash="a" * 64,
|
|
expires_at=now + timedelta(minutes=10),
|
|
created_at=now + timedelta(seconds=1),
|
|
),
|
|
]
|
|
)
|
|
await session.commit()
|
|
|
|
await email_service.reset_password_with_code(session, email, "123456", "NewPassword123")
|
|
await session.refresh(user)
|
|
assert verify_password("NewPassword123", user.password_hash)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await email_service.reset_password_with_code(session, email, "123456", "OtherPassword123")
|
|
assert exc_info.value.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_password_reset_code_verification_issues_one_time_reset_token(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
email = "verified-reset@test.com"
|
|
now = datetime.now(timezone.utc)
|
|
await add_email_settings(SessionLocal, "test.com")
|
|
async with SessionLocal() as session:
|
|
session.add(
|
|
User(
|
|
email=email,
|
|
password_hash=hash_password("OldPassword123"),
|
|
full_name="Verified Reset",
|
|
clinical_department="Clinical",
|
|
status=UserStatus.ACTIVE,
|
|
)
|
|
)
|
|
session.add(
|
|
EmailVerificationCode(
|
|
email=email,
|
|
purpose=EmailVerificationPurpose.PASSWORD_RESET,
|
|
code_hash=hash_password("654321"),
|
|
expires_at=now + timedelta(minutes=10),
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
verify_response = await client.post(
|
|
"/api/v1/auth/password-reset/email-code/verify",
|
|
json={"email": email, "code": "654321"},
|
|
)
|
|
assert verify_response.status_code == 200
|
|
assert verify_response.json()["verified"] is True
|
|
reset_token = verify_response.json()["reset_token"]
|
|
|
|
reused_code_response = await client.post(
|
|
"/api/v1/auth/password-reset/email-code/verify",
|
|
json={"email": email, "code": "654321"},
|
|
)
|
|
assert reused_code_response.status_code == 400
|
|
|
|
reset_response = await client.post(
|
|
"/api/v1/auth/password-reset-link",
|
|
json={"token": reset_token, "password": "NewPassword123"},
|
|
)
|
|
assert reset_response.status_code == 200
|
|
reused_token_response = await client.post(
|
|
"/api/v1/auth/password-reset-link",
|
|
json={"token": reset_token, "password": "OtherPassword123"},
|
|
)
|
|
assert reused_token_response.status_code == 400
|
|
|
|
async with SessionLocal() as session:
|
|
user = await user_crud.get_by_email(session, email)
|
|
assert user is not None
|
|
assert verify_password("NewPassword123", user.password_hash)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_password_reset_code_send_reports_missing_account(client_and_db):
|
|
client, _ = client_and_db
|
|
response = await client.post(
|
|
"/api/v1/auth/password-reset/email-code/send",
|
|
json={"email": "missing@test.com"},
|
|
)
|
|
assert response.status_code == 404
|
|
assert response.json()["detail"] == "该邮箱未注册"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_password_reset_link_uses_configured_frontend_url(client_and_db, monkeypatch):
|
|
client, _ = client_and_db
|
|
captured: dict[str, str] = {}
|
|
|
|
async def fake_send_password_reset_link(db, email: str, *, frontend_origin: str) -> None:
|
|
captured["email"] = email
|
|
captured["frontend_origin"] = frontend_origin
|
|
|
|
monkeypatch.setattr(email_service, "send_password_reset_link", fake_send_password_reset_link)
|
|
monkeypatch.setattr(settings, "FRONTEND_PUBLIC_URL", "https://ctms.example.com")
|
|
|
|
response = await client.post(
|
|
"/api/v1/auth/password-reset-link/send",
|
|
json={"email": "victim@test.com"},
|
|
headers={"Origin": "https://attacker.example"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert captured == {
|
|
"email": "victim@test.com",
|
|
"frontend_origin": "https://ctms.example.com",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_root_returns_service_metadata(client_and_db):
|
|
client, _ = client_and_db
|
|
resp = await client.get("/")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {
|
|
"service": "ctms-backend",
|
|
"status": "ok",
|
|
"health": "/health",
|
|
"docs": "/docs",
|
|
"api": "/api/v1",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_registered_user_can_login_after_email_verification(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
payload = {
|
|
"email": "active-register@test.com",
|
|
"password": "Password123",
|
|
"full_name": "Active Registered User",
|
|
"clinical_department": "Safety",
|
|
}
|
|
await mark_register_email_verified(SessionLocal, payload["email"])
|
|
await client.post("/api/v1/auth/register", json=payload)
|
|
resp = await encrypted_login(client, payload["email"], payload["password"])
|
|
assert resp.status_code == 200
|
|
assert resp.json()["access_token"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_created_user_is_active_by_default(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
admin_login = await encrypted_login(client, "admin@test.com", "admin123")
|
|
token = admin_login.json()["access_token"]
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
payload = {
|
|
"email": "admin-created@test.com",
|
|
"password": "Password123",
|
|
"full_name": "Admin Created",
|
|
"clinical_department": "Supply",
|
|
}
|
|
|
|
resp = await client.post("/api/v1/users/", json=payload, headers=headers)
|
|
|
|
assert resp.status_code == 201
|
|
assert resp.json()["status"] == "ACTIVE"
|
|
async with SessionLocal() as session:
|
|
user = await user_crud.get_by_email(session, payload["email"])
|
|
assert user.status == UserStatus.ACTIVE
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_users_list_filters_by_keyword_and_status(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
async with SessionLocal() as session:
|
|
session.add_all(
|
|
[
|
|
User(
|
|
email="active-filter@test.com",
|
|
password_hash=hash_password("Password123"),
|
|
full_name="Active Filter",
|
|
clinical_department="Oncology",
|
|
status=UserStatus.ACTIVE,
|
|
),
|
|
User(
|
|
email="pending-filter@test.com",
|
|
password_hash=hash_password("Password123"),
|
|
full_name="Pending Filter",
|
|
clinical_department="Cardiology",
|
|
status=UserStatus.PENDING,
|
|
),
|
|
User(
|
|
email="disabled-filter@test.com",
|
|
password_hash=hash_password("Password123"),
|
|
full_name="Disabled Filter",
|
|
clinical_department="Oncology",
|
|
status=UserStatus.DISABLED,
|
|
),
|
|
]
|
|
)
|
|
await session.commit()
|
|
|
|
admin_login = await encrypted_login(client, "admin@test.com", "admin123")
|
|
token = admin_login.json()["access_token"]
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
active_resp = await client.get("/api/v1/users/?status=ACTIVE", headers=headers)
|
|
assert active_resp.status_code == 200
|
|
active_items = active_resp.json()["items"]
|
|
assert all(item["status"] == "ACTIVE" for item in active_items)
|
|
assert {item["email"] for item in active_items} >= {"admin@test.com", "active-filter@test.com"}
|
|
assert "pending-filter@test.com" not in {item["email"] for item in active_items}
|
|
|
|
keyword_resp = await client.get("/api/v1/users/?keyword=Oncology", headers=headers)
|
|
assert keyword_resp.status_code == 200
|
|
keyword_data = keyword_resp.json()
|
|
assert keyword_data["total"] == 2
|
|
assert {item["email"] for item in keyword_data["items"]} == {
|
|
"active-filter@test.com",
|
|
"disabled-filter@test.com",
|
|
}
|
|
|
|
combined_resp = await client.get("/api/v1/users/?keyword=Filter&status=PENDING", headers=headers)
|
|
assert combined_resp.status_code == 200
|
|
combined_data = combined_resp.json()
|
|
assert combined_data["total"] == 1
|
|
assert combined_data["items"][0]["email"] == "pending-filter@test.com"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_delete_user_with_audit_history_returns_400(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
async with SessionLocal() as session:
|
|
user = User(
|
|
email="audited-delete@test.com",
|
|
password_hash=hash_password("Password123"),
|
|
full_name="Audited Delete",
|
|
clinical_department="Medical",
|
|
status=UserStatus.ACTIVE,
|
|
)
|
|
session.add(user)
|
|
await session.flush()
|
|
session.add(
|
|
AuditLog(
|
|
entity_type="user",
|
|
entity_id=user.id,
|
|
action="LOGIN",
|
|
operator_id=user.id,
|
|
operator_role="USER",
|
|
)
|
|
)
|
|
await session.commit()
|
|
user_id = user.id
|
|
|
|
admin_login = await encrypted_login(client, "admin@test.com", "admin123")
|
|
token = admin_login.json()["access_token"]
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
resp = await client.delete(f"/api/v1/users/{user_id}", headers=headers)
|
|
|
|
assert resp.status_code == 400
|
|
assert resp.json()["detail"] == "该账号已有审计或权限访问记录，请停用账号以保留历史追溯"
|
|
async with SessionLocal() as session:
|
|
assert await user_crud.get_by_id(session, user_id) is not None
|
|
|
|
|
|
def test_register_request_does_not_expose_role_input():
|
|
assert "role" not in UserRegisterRequest.model_fields
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plaintext_login_is_rejected(client_and_db):
|
|
client, _ = client_and_db
|
|
resp = await client.post("/api/v1/auth/login", json={"email": "admin@test.com", "password": "admin123"})
|
|
assert resp.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dev_login_allows_plaintext_only_in_development(client_and_db):
|
|
client, _ = client_and_db
|
|
original_env = settings.ENV
|
|
settings.ENV = "development"
|
|
try:
|
|
resp = await client.post("/api/v1/auth/dev-login", json={"email": "admin@test.com", "password": "admin123"})
|
|
finally:
|
|
settings.ENV = original_env
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["token_type"] == "bearer"
|
|
assert resp.json()["access_token"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_avatar_upload_rejects_non_image_files(client_and_db):
|
|
client, _ = client_and_db
|
|
original_env = settings.ENV
|
|
settings.ENV = "development"
|
|
try:
|
|
login_resp = await client.post("/api/v1/auth/dev-login", json={"email": "admin@test.com", "password": "admin123"})
|
|
finally:
|
|
settings.ENV = original_env
|
|
|
|
token = login_resp.json()["access_token"]
|
|
resp = await client.post(
|
|
"/api/v1/auth/me/avatar",
|
|
files={"file": ("avatar.txt", b"not an image", "text/plain")},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
|
|
assert resp.status_code == 400
|
|
assert resp.json()["detail"] == "头像仅支持图片格式"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dev_login_is_disabled_outside_development(client_and_db):
|
|
client, _ = client_and_db
|
|
original_env = settings.ENV
|
|
settings.ENV = "production"
|
|
try:
|
|
resp = await client.post("/api/v1/auth/dev-login", json={"email": "admin@test.com", "password": "admin123"})
|
|
finally:
|
|
settings.ENV = original_env
|
|
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dev_login_rejects_inactive_users(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
original_env = settings.ENV
|
|
settings.ENV = "development"
|
|
try:
|
|
async with SessionLocal() as session:
|
|
session.add(
|
|
User(
|
|
email="pending-dev-login@test.com",
|
|
password_hash=hash_password("Password123"),
|
|
full_name="Pending Dev Login",
|
|
clinical_department="Clinical",
|
|
status=UserStatus.PENDING,
|
|
)
|
|
)
|
|
await session.commit()
|
|
resp = await client.post(
|
|
"/api/v1/auth/dev-login",
|
|
json={"email": "pending-dev-login@test.com", "password": "Password123"},
|
|
)
|
|
finally:
|
|
settings.ENV = original_env
|
|
|
|
assert resp.status_code == 401
|
|
assert "账号未审核" in resp.json().get("detail", "")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_challenge_cannot_be_reused(client_and_db):
|
|
client, _ = client_and_db
|
|
key_resp = await client.get("/api/v1/auth/login-key")
|
|
login_key = key_resp.json()
|
|
public_key = serialization.load_pem_public_key(login_key["public_key"].encode("utf-8"))
|
|
plaintext = json.dumps(
|
|
{
|
|
"email": "admin@test.com",
|
|
"password": "admin123",
|
|
"challenge": login_key["challenge"],
|
|
},
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
aes_key = AESGCM.generate_key(bit_length=256)
|
|
iv = os.urandom(12)
|
|
encrypted_data = AESGCM(aes_key).encrypt(iv, plaintext, None)
|
|
encrypted_key = public_key.encrypt(
|
|
aes_key,
|
|
padding.OAEP(
|
|
mgf=padding.MGF1(algorithm=hashes.SHA256()),
|
|
algorithm=hashes.SHA256(),
|
|
label=None,
|
|
)
|
|
)
|
|
ciphertext = base64.b64encode(
|
|
json.dumps(
|
|
{
|
|
"encrypted_key": base64.b64encode(encrypted_key).decode("ascii"),
|
|
"iv": base64.b64encode(iv).decode("ascii"),
|
|
"data": base64.b64encode(encrypted_data).decode("ascii"),
|
|
},
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
).decode("ascii")
|
|
payload = {
|
|
"key_id": login_key["key_id"],
|
|
"challenge": login_key["challenge"],
|
|
"ciphertext": ciphertext,
|
|
}
|
|
|
|
first = await client.post("/api/v1/auth/login", json=payload)
|
|
second = await client.post("/api/v1/auth/login", json=payload)
|
|
|
|
assert first.status_code == 200
|
|
assert second.status_code == 401
|