security(07.3-02): ES256 signing — swap 4 JWT sites + config keys + promote tests
- backend/config.py: add refresh_token_expire_hours=16, jwt_private_key, jwt_public_key - backend/services/auth.py: add import base64; all 4 JWT sites use ES256 with inline PEM decode; zero HS256/secret_key references - backend/tests/test_auth_es256.py: promote ES256-01..03 + CFG-01 tests from xfail to passing
This commit is contained in:
@@ -33,6 +33,10 @@ class Settings(BaseSettings):
|
|||||||
# Auth / JWT (Phase 2)
|
# Auth / JWT (Phase 2)
|
||||||
access_token_expire_minutes: int = 15
|
access_token_expire_minutes: int = 15
|
||||||
refresh_token_expire_days: int = 30
|
refresh_token_expire_days: int = 30
|
||||||
|
# Phase 7.3 — D-01, D-09: ES256 key pair + short-session TTL
|
||||||
|
refresh_token_expire_hours: int = 16 # default short session
|
||||||
|
jwt_private_key: str = "" # base64-encoded PEM; required in production
|
||||||
|
jwt_public_key: str = "" # base64-encoded PEM; required in production
|
||||||
|
|
||||||
# SMTP (Phase 2 — D-01)
|
# SMTP (Phase 2 — D-01)
|
||||||
smtp_host: str = ""
|
smtp_host: str = ""
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ Security invariants:
|
|||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import base64
|
||||||
import hashlib
|
import hashlib
|
||||||
import hmac
|
import hmac
|
||||||
import logging
|
import logging
|
||||||
@@ -97,7 +98,8 @@ def create_access_token(user_id: str, role: str) -> str:
|
|||||||
"exp": now + timedelta(minutes=settings.access_token_expire_minutes),
|
"exp": now + timedelta(minutes=settings.access_token_expire_minutes),
|
||||||
"jti": str(uuid.uuid4()),
|
"jti": str(uuid.uuid4()),
|
||||||
}
|
}
|
||||||
return jwt.encode(payload, settings.secret_key, algorithm="HS256")
|
private_pem = base64.b64decode(settings.jwt_private_key).decode()
|
||||||
|
return jwt.encode(payload, private_pem, algorithm="ES256")
|
||||||
|
|
||||||
|
|
||||||
def decode_access_token(token: str) -> dict:
|
def decode_access_token(token: str) -> dict:
|
||||||
@@ -107,7 +109,8 @@ def decode_access_token(token: str) -> dict:
|
|||||||
tokens from being used as access tokens).
|
tokens from being used as access tokens).
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
|
public_pem = base64.b64decode(settings.jwt_public_key).decode()
|
||||||
|
payload = jwt.decode(token, public_pem, algorithms=["ES256"])
|
||||||
except jwt.ExpiredSignatureError as exc:
|
except jwt.ExpiredSignatureError as exc:
|
||||||
raise ValueError("Token has expired") from exc
|
raise ValueError("Token has expired") from exc
|
||||||
except jwt.PyJWTError as exc:
|
except jwt.PyJWTError as exc:
|
||||||
@@ -130,7 +133,8 @@ def create_password_reset_token(user_id: str) -> str:
|
|||||||
"iat": now,
|
"iat": now,
|
||||||
"exp": now + timedelta(seconds=3600),
|
"exp": now + timedelta(seconds=3600),
|
||||||
}
|
}
|
||||||
return jwt.encode(payload, settings.secret_key, algorithm="HS256")
|
private_pem = base64.b64decode(settings.jwt_private_key).decode()
|
||||||
|
return jwt.encode(payload, private_pem, algorithm="ES256")
|
||||||
|
|
||||||
|
|
||||||
def decode_password_reset_token(token: str) -> str:
|
def decode_password_reset_token(token: str) -> str:
|
||||||
@@ -139,7 +143,8 @@ def decode_password_reset_token(token: str) -> str:
|
|||||||
Returns the user_id string.
|
Returns the user_id string.
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
|
public_pem = base64.b64decode(settings.jwt_public_key).decode()
|
||||||
|
payload = jwt.decode(token, public_pem, algorithms=["ES256"])
|
||||||
except jwt.ExpiredSignatureError as exc:
|
except jwt.ExpiredSignatureError as exc:
|
||||||
raise ValueError("Reset token has expired") from exc
|
raise ValueError("Reset token has expired") from exc
|
||||||
except jwt.PyJWTError as exc:
|
except jwt.PyJWTError as exc:
|
||||||
|
|||||||
@@ -36,21 +36,47 @@ def es256_keys(monkeypatch):
|
|||||||
monkeypatch.setattr("config.settings.jwt_public_key", pub, raising=False)
|
monkeypatch.setattr("config.settings.jwt_public_key", pub, raising=False)
|
||||||
|
|
||||||
|
|
||||||
# ── ES256 algorithm stubs ─────────────────────────────────────────────────────
|
# ── ES256 algorithm tests ─────────────────────────────────────────────────────
|
||||||
|
|
||||||
@pytest.mark.xfail(strict=True, reason="ES256-01: not yet implemented")
|
|
||||||
def test_access_token_uses_es256():
|
def test_access_token_uses_es256():
|
||||||
pytest.xfail("not yet implemented")
|
"""ES256-01: access token header must declare alg=ES256."""
|
||||||
|
from services.auth import create_access_token
|
||||||
|
token = create_access_token("u1", "user")
|
||||||
|
# Decode the header (first segment of the JWT)
|
||||||
|
header_b64 = token.split(".")[0]
|
||||||
|
# Add padding to make it valid base64
|
||||||
|
padding = "=" * (4 - len(header_b64) % 4)
|
||||||
|
header = json.loads(base64.urlsafe_b64decode(header_b64 + padding))
|
||||||
|
assert header["alg"] == "ES256"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(strict=True, reason="ES256-02: not yet implemented")
|
|
||||||
def test_hs256_token_rejected():
|
def test_hs256_token_rejected():
|
||||||
pytest.xfail("not yet implemented")
|
"""ES256-02: an HS256 token must raise ValueError when decoded."""
|
||||||
|
import jwt as _jwt
|
||||||
|
from services.auth import decode_access_token
|
||||||
|
hs256_token = _jwt.encode(
|
||||||
|
{
|
||||||
|
"sub": "u1",
|
||||||
|
"typ": "access",
|
||||||
|
"exp": int(time.time()) + 60,
|
||||||
|
"iat": int(time.time()),
|
||||||
|
},
|
||||||
|
"any-hs256-secret",
|
||||||
|
algorithm="HS256",
|
||||||
|
)
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
decode_access_token(hs256_token)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(strict=True, reason="ES256-03: not yet implemented")
|
|
||||||
def test_reset_token_uses_es256():
|
def test_reset_token_uses_es256():
|
||||||
pytest.xfail("not yet implemented")
|
"""ES256-03: password-reset token header must declare alg=ES256 and round-trips."""
|
||||||
|
from services.auth import create_password_reset_token, decode_password_reset_token
|
||||||
|
token = create_password_reset_token("u1")
|
||||||
|
header_b64 = token.split(".")[0]
|
||||||
|
padding = "=" * (4 - len(header_b64) % 4)
|
||||||
|
header = json.loads(base64.urlsafe_b64decode(header_b64 + padding))
|
||||||
|
assert header["alg"] == "ES256"
|
||||||
|
assert decode_password_reset_token(token) == "u1"
|
||||||
|
|
||||||
|
|
||||||
# ── Startup token rotation stubs ──────────────────────────────────────────────
|
# ── Startup token rotation stubs ──────────────────────────────────────────────
|
||||||
@@ -87,8 +113,12 @@ async def test_remember_me_cookie_max_age(async_client, auth_user):
|
|||||||
pytest.xfail("not yet implemented")
|
pytest.xfail("not yet implemented")
|
||||||
|
|
||||||
|
|
||||||
# ── Config field stubs ────────────────────────────────────────────────────────
|
# ── Config field tests ────────────────────────────────────────────────────────
|
||||||
|
|
||||||
@pytest.mark.xfail(strict=True, reason="CFG-01: not yet implemented")
|
|
||||||
def test_settings_has_jwt_keys():
|
def test_settings_has_jwt_keys():
|
||||||
pytest.xfail("not yet implemented")
|
"""CFG-01: Settings must expose jwt_private_key, jwt_public_key, and refresh_token_expire_hours=16."""
|
||||||
|
from config import settings
|
||||||
|
assert hasattr(settings, "jwt_private_key")
|
||||||
|
assert hasattr(settings, "jwt_public_key")
|
||||||
|
assert hasattr(settings, "refresh_token_expire_hours")
|
||||||
|
assert settings.refresh_token_expire_hours == 16
|
||||||
|
|||||||
Reference in New Issue
Block a user