feat: harden auth and study workflows
- replace plaintext login and unlock requests with RSA-OAEP/AES-GCM encrypted payloads - add login challenge replay protection, production RSA key validation, and auth tests - wire compose to environment-driven dev/prod settings without committing local secrets - update setup-config smoke scripts and Postman docs for encrypted login - add visit schedule migrations/tests and update study/subject setup workflows
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
import asyncio
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
@@ -6,6 +7,9 @@ import urllib.error
|
||||
import urllib.request
|
||||
|
||||
import asyncpg
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import padding
|
||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||
|
||||
|
||||
BASE = os.getenv("BASE_URL", "http://localhost:8000")
|
||||
@@ -66,13 +70,53 @@ def db_fetch(sql: str, *args) -> list[dict]:
|
||||
return asyncio.run(_db_fetch(sql, *args))
|
||||
|
||||
|
||||
def main() -> int:
|
||||
print(f"[config] BASE={BASE} EMAIL={ADMIN_EMAIL}")
|
||||
status, login = request_json(
|
||||
def encrypted_login(email: str, password: str) -> tuple[int, dict]:
|
||||
status, login_key = request_json("/api/v1/auth/login-key")
|
||||
if status != 200:
|
||||
return status, login_key
|
||||
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 request_json(
|
||||
"/api/v1/auth/login",
|
||||
method="POST",
|
||||
payload={"email": ADMIN_EMAIL, "password": ADMIN_PASSWORD},
|
||||
payload={
|
||||
"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"),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
print(f"[config] BASE={BASE} EMAIL={ADMIN_EMAIL}")
|
||||
status, login = encrypted_login(ADMIN_EMAIL, ADMIN_PASSWORD)
|
||||
assert_or_exit(status == 200, f"登录失败 status={status} body={login}")
|
||||
token = login["access_token"]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user