feat(backend): pose les primitives de sécurité

Couche pure, sans FastAPI ni session : rôles ordonnés, `Principal`,
encodage et décodage des jetons d'accès, empreinte des jetons de
rafraîchissement, et hachage Argon2id poussé dans un fil borné.

Aucun de ces modules ne lit `get_settings()`, mis en cache par
`lru_cache` et donc contaminé entre tests : les paramètres arrivent par
`TokenPolicy` et par `build_hasher()`.

Argon2id est calibré à m=19456 KiB, t=2, p=1, soit 17 ms mesurés sur un
poste de développement.
This commit is contained in:
Johan LEROY
2026-09-15 14:25:41 +02:00
parent 008cf581a7
commit 53af7a76d8
9 changed files with 627 additions and 1 deletions
+61
View File
@@ -0,0 +1,61 @@
from app.core.hashing import Argon2Hasher, build_hasher
MOT_DE_PASSE = "un-mot-de-passe-de-test-assez-long"
def fabrique(time_cost: int = 1, max_concurrency: int = 2) -> Argon2Hasher:
return build_hasher(
time_cost=time_cost,
memory_cost_kib=8192,
parallelism=1,
max_concurrency=max_concurrency,
)
async def test_hash_produces_a_distinct_digest_for_the_same_password() -> None:
hacheur = fabrique()
premier = await hacheur.hash(MOT_DE_PASSE)
second = await hacheur.hash(MOT_DE_PASSE)
assert premier != second
assert premier.startswith("$argon2id$")
async def test_verify_accepts_the_right_password_and_rejects_the_others() -> None:
hacheur = fabrique()
empreinte = await hacheur.hash(MOT_DE_PASSE)
assert await hacheur.verify(empreinte, MOT_DE_PASSE) is True
assert await hacheur.verify(empreinte, "un-autre-mot-de-passe") is False
async def test_verify_returns_false_when_the_stored_digest_is_malformed() -> None:
hacheur = fabrique()
accorde = await hacheur.verify("pas-une-empreinte-argon2", MOT_DE_PASSE)
assert accorde is False
async def test_needs_rehash_is_true_when_the_parameters_changed() -> None:
ancien = fabrique(time_cost=1)
recent = fabrique(time_cost=3)
empreinte = await ancien.hash(MOT_DE_PASSE)
assert ancien.needs_rehash(empreinte) is False
assert recent.needs_rehash(empreinte) is True
def test_needs_rehash_is_true_when_the_stored_digest_is_malformed() -> None:
hacheur = fabrique()
assert hacheur.needs_rehash("pas-une-empreinte-argon2") is True
async def test_verify_dummy_completes_without_revealing_anything() -> None:
hacheur = fabrique()
await hacheur.verify_dummy()
+40
View File
@@ -0,0 +1,40 @@
import pytest
from app.core.roles import Role, has_at_least
@pytest.mark.parametrize(
("actual", "required", "expected"),
[
(Role.LECTEUR, Role.LECTEUR, True),
(Role.LECTEUR, Role.OPERATEUR, False),
(Role.LECTEUR, Role.ADMIN, False),
(Role.OPERATEUR, Role.LECTEUR, True),
(Role.OPERATEUR, Role.OPERATEUR, True),
(Role.OPERATEUR, Role.ADMIN, False),
(Role.ADMIN, Role.LECTEUR, True),
(Role.ADMIN, Role.OPERATEUR, True),
(Role.ADMIN, Role.ADMIN, True),
],
ids=[
"lecteur_sur_lecteur",
"lecteur_sur_operateur",
"lecteur_sur_admin",
"operateur_sur_lecteur",
"operateur_sur_operateur",
"operateur_sur_admin",
"admin_sur_lecteur",
"admin_sur_operateur",
"admin_sur_admin",
],
)
def test_has_at_least_orders_the_three_roles(actual: Role, required: Role, expected: bool) -> None:
accorde = has_at_least(actual, required)
assert accorde is expected
def test_role_values_stay_ascii_for_the_wire_format() -> None:
valeurs = [role.value for role in Role]
assert all(valeur.isascii() for valeur in valeurs)
+194
View File
@@ -0,0 +1,194 @@
import base64
import json
from datetime import UTC, datetime, timedelta
from uuid import uuid4
import jwt
import pytest
from app.core.security import (
AccessClaims,
TokenExpiredError,
TokenInvalidError,
TokenPolicy,
decode_access_token,
encode_access_token,
fingerprint_refresh,
generate_refresh_secret,
)
POLITIQUE = TokenPolicy(
secret="un-secret-de-test-de-plus-de-trente-deux-caracteres",
issuer="enervision-api",
audience="enervision-web",
access_ttl=timedelta(minutes=15),
)
def emets(**surcharges: object) -> str:
charge = {
"iss": POLITIQUE.issuer,
"aud": POLITIQUE.audience,
"sub": str(uuid4()),
"iat": datetime.now(UTC),
"exp": datetime.now(UTC) + timedelta(minutes=15),
"jti": str(uuid4()),
"typ": "access",
"role": "lecteur",
"kind": "human",
}
charge.update(surcharges)
return jwt.encode(charge, POLITIQUE.secret, algorithm="HS256")
def test_decode_access_token_returns_the_claims_when_the_token_is_valid() -> None:
sujet = uuid4()
jeton = encode_access_token(POLITIQUE, subject=sujet, role="operateur", kind="human")
claims = decode_access_token(POLITIQUE, jeton)
assert isinstance(claims, AccessClaims)
assert claims.subject == sujet
assert claims.role == "operateur"
assert claims.kind == "human"
def test_decode_access_token_raises_expired_when_the_lifetime_has_passed() -> None:
passe = datetime.now(UTC) - timedelta(hours=2)
jeton = encode_access_token(POLITIQUE, subject=uuid4(), role="lecteur", kind="human", now=passe)
with pytest.raises(TokenExpiredError):
decode_access_token(POLITIQUE, jeton)
def test_decode_access_token_raises_invalid_when_the_signature_was_forged() -> None:
autre = TokenPolicy(
secret="un-autre-secret-tout-aussi-long-que-le-premier",
issuer=POLITIQUE.issuer,
audience=POLITIQUE.audience,
access_ttl=POLITIQUE.access_ttl,
)
jeton = encode_access_token(autre, subject=uuid4(), role="lecteur", kind="human")
with pytest.raises(TokenInvalidError):
decode_access_token(POLITIQUE, jeton)
@pytest.mark.parametrize(
"surcharges",
[
{"aud": "un-autre-public"},
{"iss": "un-autre-emetteur"},
{"typ": "refresh"},
],
ids=["audience_invalide", "emetteur_invalide", "jeton_de_rafraichissement"],
)
def test_decode_access_token_raises_invalid_when_a_claim_is_wrong(
surcharges: dict[str, object],
) -> None:
jeton = emets(**surcharges)
with pytest.raises(TokenInvalidError):
decode_access_token(POLITIQUE, jeton)
@pytest.mark.parametrize(
"claim",
["jti", "typ", "role", "kind"],
ids=["identifiant", "type", "role", "nature_du_compte"],
)
def test_decode_access_token_raises_invalid_when_a_required_claim_is_missing(claim: str) -> None:
charge = {
"iss": POLITIQUE.issuer,
"aud": POLITIQUE.audience,
"sub": str(uuid4()),
"iat": datetime.now(UTC),
"exp": datetime.now(UTC) + timedelta(minutes=15),
"jti": str(uuid4()),
"typ": "access",
"role": "lecteur",
"kind": "human",
}
del charge[claim]
jeton = jwt.encode(charge, POLITIQUE.secret, algorithm="HS256")
with pytest.raises(TokenInvalidError):
decode_access_token(POLITIQUE, jeton)
def test_decode_access_token_rejects_a_token_forged_with_the_none_algorithm() -> None:
def encode(donnees: dict[str, object]) -> str:
brut = json.dumps(donnees, separators=(",", ":")).encode()
return base64.urlsafe_b64encode(brut).rstrip(b"=").decode()
entete = encode({"alg": "none", "typ": "JWT"})
charge = encode(
{
"iss": POLITIQUE.issuer,
"aud": POLITIQUE.audience,
"sub": str(uuid4()),
"iat": int(datetime.now(UTC).timestamp()),
"exp": int((datetime.now(UTC) + timedelta(minutes=15)).timestamp()),
"jti": str(uuid4()),
"typ": "access",
"role": "admin",
"kind": "human",
}
)
with pytest.raises(TokenInvalidError):
decode_access_token(POLITIQUE, f"{entete}.{charge}.")
def test_decode_access_token_rejects_a_token_signed_with_another_algorithm() -> None:
charge = {
"iss": POLITIQUE.issuer,
"aud": POLITIQUE.audience,
"sub": str(uuid4()),
"iat": datetime.now(UTC),
"exp": datetime.now(UTC) + timedelta(minutes=15),
"jti": str(uuid4()),
"typ": "access",
"role": "admin",
"kind": "human",
}
jeton = jwt.encode(charge, POLITIQUE.secret * 2, algorithm="HS512")
with pytest.raises(TokenInvalidError):
decode_access_token(POLITIQUE, jeton)
@pytest.mark.parametrize(
"surcharges",
[{"sub": "pas-un-uuid"}, {"jti": "pas-un-uuid"}],
ids=["sujet_illisible", "identifiant_illisible"],
)
def test_decode_access_token_raises_invalid_when_an_identifier_is_not_a_uuid(
surcharges: dict[str, object],
) -> None:
jeton = emets(**surcharges)
with pytest.raises(TokenInvalidError):
decode_access_token(POLITIQUE, jeton)
def test_generate_refresh_secret_returns_distinct_url_safe_values() -> None:
secrets_generes = {generate_refresh_secret() for _ in range(100)}
assert len(secrets_generes) == 100
assert all(len(valeur) >= 43 for valeur in secrets_generes)
def test_fingerprint_refresh_is_stable_and_distinguishes_two_secrets() -> None:
premier = generate_refresh_secret()
second = generate_refresh_secret()
empreinte = fingerprint_refresh(premier)
assert len(empreinte) == 32
assert empreinte == fingerprint_refresh(premier)
assert empreinte != fingerprint_refresh(second)