|
| 1 | +"""Edge-case JWT handling around the security manager's permission check.""" |
| 2 | + |
| 3 | +from datetime import datetime, timedelta, timezone |
| 4 | + |
| 5 | +import jwt |
| 6 | + |
| 7 | +from security.manager import is_permitted, load_legacy_records |
| 8 | +from server.auth import ( |
| 9 | + ROLE_ADMIN, |
| 10 | + _jwt_algorithm, |
| 11 | + _jwt_secret, |
| 12 | + create_access_token, |
| 13 | +) |
| 14 | + |
| 15 | + |
| 16 | +COUNTRIES_PROTOCOL = { |
| 17 | + "countries": { |
| 18 | + "create": { |
| 19 | + "checks": { |
| 20 | + "login": True, |
| 21 | + "allowed_roles": [ROLE_ADMIN], |
| 22 | + }, |
| 23 | + }, |
| 24 | + }, |
| 25 | +} |
| 26 | + |
| 27 | + |
| 28 | +def _expired_admin_token() -> str: |
| 29 | + now = datetime.now(timezone.utc) |
| 30 | + payload = { |
| 31 | + "sub": "alice", |
| 32 | + "role": ROLE_ADMIN, |
| 33 | + "iat": now - timedelta(hours=2), |
| 34 | + "exp": now - timedelta(hours=1), |
| 35 | + } |
| 36 | + return jwt.encode(payload, _jwt_secret(), algorithm=_jwt_algorithm()) |
| 37 | + |
| 38 | + |
| 39 | +def _wrong_secret_admin_token() -> str: |
| 40 | + now = datetime.now(timezone.utc) |
| 41 | + payload = { |
| 42 | + "sub": "alice", |
| 43 | + "role": ROLE_ADMIN, |
| 44 | + "iat": now, |
| 45 | + "exp": now + timedelta(hours=1), |
| 46 | + } |
| 47 | + return jwt.encode( |
| 48 | + payload, "definitely-not-the-real-secret", algorithm=_jwt_algorithm() |
| 49 | + ) |
| 50 | + |
| 51 | + |
| 52 | +def test_malformed_bearer_token_is_denied(): |
| 53 | + load_legacy_records(COUNTRIES_PROTOCOL) |
| 54 | + assert not is_permitted( |
| 55 | + "countries", "create", auth_header="Bearer not-a-jwt" |
| 56 | + ) |
| 57 | + |
| 58 | + |
| 59 | +def test_empty_bearer_token_is_denied(): |
| 60 | + load_legacy_records(COUNTRIES_PROTOCOL) |
| 61 | + assert not is_permitted( |
| 62 | + "countries", "create", auth_header="Bearer " |
| 63 | + ) |
| 64 | + |
| 65 | + |
| 66 | +def test_non_bearer_authorization_scheme_is_denied(): |
| 67 | + load_legacy_records(COUNTRIES_PROTOCOL) |
| 68 | + assert not is_permitted( |
| 69 | + "countries", "create", auth_header="Basic dXNlcjpwYXNz" |
| 70 | + ) |
| 71 | + |
| 72 | + |
| 73 | +def test_expired_admin_token_is_denied(): |
| 74 | + load_legacy_records(COUNTRIES_PROTOCOL) |
| 75 | + assert not is_permitted( |
| 76 | + "countries", |
| 77 | + "create", |
| 78 | + auth_header=f"Bearer {_expired_admin_token()}", |
| 79 | + ) |
| 80 | + |
| 81 | + |
| 82 | +def test_token_signed_with_wrong_secret_is_denied(): |
| 83 | + load_legacy_records(COUNTRIES_PROTOCOL) |
| 84 | + assert not is_permitted( |
| 85 | + "countries", |
| 86 | + "create", |
| 87 | + auth_header=f"Bearer {_wrong_secret_admin_token()}", |
| 88 | + ) |
| 89 | + |
| 90 | + |
| 91 | +def test_valid_admin_token_is_still_permitted(): |
| 92 | + load_legacy_records(COUNTRIES_PROTOCOL) |
| 93 | + token = create_access_token("alice", ROLE_ADMIN, expires_hours=1) |
| 94 | + assert is_permitted( |
| 95 | + "countries", |
| 96 | + "create", |
| 97 | + auth_header=f"Bearer {token}", |
| 98 | + ) |
0 commit comments