Coverage for src/lexigram/admin/auth/services/csrf_service.py: 84%

1"""Admin CSRF token generation and validation service.""" 

2 

3from __future__ import annotations 

4 

5import base64 

6import hashlib 

7import hmac 

8import secrets 

9import time 

10 

11from lexigram.admin.auth.protocols import AdminCsrfServiceProtocol 

12from lexigram.di.decorators import inject 

13from lexigram.logging import get_logger 

14 

15logger = get_logger(__name__) 

16 

17# Token wire format (before base64url encoding): 

18# "{timestamp}:{nonce}:{hmac_signature}" 

19# 

20# HMAC message: f"{session_id}:{timestamp}:{nonce}" 

21# HMAC key: sha256(secret.encode()) — derived once at construction time 

22# HMAC digest: SHA-256 hexdigest 

23 

24 

25@inject 

26class AdminCsrfService: 

27 """CSRF token generation and validation service. 

28 

29 Tokens are HMAC-SHA256 signed, scoped to a specific session ID, 

30 and expire after a configurable lifetime (default 1 hour). 

31 Validation uses ``hmac.compare_digest`` for timing-safe comparison. 

32 """ 

33 

34 def __init__( 

35 self, 

36 secret: str, 

37 token_lifetime: int = 3600, 

38 ) -> None: 

39 """Initialize with signing secret. 

40 

41 Args: 

42 secret: HMAC signing secret (use AdminAuthConfig.session_secret). 

43 token_lifetime: Token validity in seconds (default 3600 = 1 hour). 

44 """ 

45 # Derive a fixed-length key from the caller-supplied secret so that 

46 # the key is always exactly 32 bytes regardless of secret length. 

47 self._key: bytes = hashlib.sha256(secret.encode()).digest() 

48 self._lifetime: int = token_lifetime 

49 

50 @property 

51 def token_lifetime_seconds(self) -> int: 

52 """Return the configured token lifetime in seconds.""" 

53 return self._lifetime 

54 

55 # ------------------------------------------------------------------ 

56 # AdminCsrfServiceProtocol 

57 # ------------------------------------------------------------------ 

58 

59 def generate_token(self, session_id: str) -> str: 

60 """Generate a CSRF token scoped to the given session. 

61 

62 Token format: ``base64url("{timestamp}:{nonce}:{hmac_signature}")``. 

63 

64 Args: 

65 session_id: Session ID to scope this token to. 

66 

67 Returns: 

68 URL-safe base64-encoded CSRF token string. 

69 """ 

70 timestamp = str(int(time.time())) 

71 nonce = secrets.token_hex(16) 

72 message = f"{session_id}:{timestamp}:{nonce}" 

73 signature = hmac.new(self._key, message.encode(), hashlib.sha256).hexdigest() 

74 raw = f"{timestamp}:{nonce}:{signature}" 

75 token = base64.urlsafe_b64encode(raw.encode()).decode() 

76 logger.debug("csrf.token_generated", session_id=session_id) 

77 return token 

78 

79 def validate_token(self, session_id: str, token: str) -> bool: 

80 """Validate a CSRF token against the session. 

81 

82 Uses ``hmac.compare_digest`` for timing-safe comparison. 

83 

84 Args: 

85 session_id: Session ID the token was generated for. 

86 token: Token to validate (base64url-encoded). 

87 

88 Returns: 

89 True if the token is valid and not expired, False otherwise. 

90 """ 

91 try: 

92 raw = base64.urlsafe_b64decode(token.encode()).decode() 

93 parts = raw.split(":", 2) 

94 if len(parts) != 3: 

95 logger.debug("csrf.token_malformed", session_id=session_id) 

96 return False 

97 

98 timestamp_str, nonce, provided_signature = parts 

99 

100 # Expiry check 

101 token_age = int(time.time()) - int(timestamp_str) 

102 if token_age > self._lifetime: 

103 logger.debug( 

104 "csrf.token_expired", 

105 session_id=session_id, 

106 age_seconds=token_age, 

107 ) 

108 return False 

109 

110 # Recompute and compare in constant time 

111 message = f"{session_id}:{timestamp_str}:{nonce}" 

112 expected_signature = hmac.new( 

113 self._key, message.encode(), hashlib.sha256 

114 ).hexdigest() 

115 

116 valid = hmac.compare_digest(expected_signature, provided_signature) 

117 if not valid: 

118 logger.debug("csrf.token_signature_mismatch", session_id=session_id) 

119 return valid 

120 

121 except (ValueError, TypeError) as exc: 

122 logger.debug( 

123 "csrf.token_validation_error", 

124 session_id=session_id, 

125 error=str(exc), 

126 ) 

127 return False 

128 

129 

130# Verify that the concrete class satisfies the protocol at import time. 

131_: AdminCsrfServiceProtocol = AdminCsrfService.__new__(AdminCsrfService) 

132 

133__all__ = ["AdminCsrfService"]