Coverage for src / lexigram / contracts / security / protocols.py: 0%

1"""Security protocol definitions for the Lexigram Framework. 

2 

3Protocols for guard chains, input sanitization, and security header management. 

4All security implementations must satisfy these structural interfaces. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

10 

11if TYPE_CHECKING: 

12 from lexigram.contracts.web.guard import GuardProtocol 

13 

14 

15@runtime_checkable 

16class HasherProtocol(Protocol): 

17 """General-purpose hashing protocol (non-password-specific).""" 

18 

19 @property 

20 def algorithm(self) -> str: 

21 """Hash algorithm name (for example ``sha256`` or ``blake2b``).""" 

22 ... 

23 

24 def digest(self, data: str | bytes) -> str: 

25 """Hash input data and return an encoded digest. 

26 

27 Args: 

28 data: Input string or bytes to hash. 

29 

30 Returns: 

31 Encoded hash string. 

32 """ 

33 ... 

34 

35 def verify_digest(self, data: str | bytes, expected: str) -> bool: 

36 """Constant-time verification against an expected digest. 

37 

38 Args: 

39 data: Input string or bytes to hash. 

40 expected: Expected encoded digest. 

41 

42 Returns: 

43 True when the digest matches, False otherwise. 

44 """ 

45 ... 

46 

47 async def hash(self, value: str) -> str: 

48 """Backward-compatible async alias for :meth:`digest`.""" 

49 ... 

50 

51 async def verify(self, value: str, hashed_value: str) -> bool: 

52 """Backward-compatible async alias for :meth:`verify_digest`.""" 

53 ... 

54 

55 

56@runtime_checkable 

57class KeyDerivationProtocol(Protocol): 

58 """Protocol for key derivation services.""" 

59 

60 async def derive(self, secret: str, *, salt: bytes | None = None) -> str: 

61 """Derive an encoded key from a secret. 

62 

63 Args: 

64 secret: Input secret to derive from. 

65 salt: Optional salt. When omitted, implementations generate one. 

66 

67 Returns: 

68 Stable encoded key derivation string. 

69 """ 

70 ... 

71 

72 async def verify(self, secret: str, encoded: str) -> bool: 

73 """Verify a secret against a derived key payload. 

74 

75 Args: 

76 secret: Input secret to verify. 

77 encoded: Stable encoded key derivation string. 

78 

79 Returns: 

80 True when the secret matches the encoded payload, False otherwise. 

81 """ 

82 ... 

83 

84 async def hash(self, secret: str, *, salt: bytes | None = None) -> str: 

85 """Backward-compatible async alias for :meth:`derive`.""" 

86 ... 

87 

88 

89@runtime_checkable 

90class GuardChainProtocol(Protocol): 

91 """Executes a sequence of guards; short-circuits on first denial. 

92 

93 Implementations must raise ``GuardDeniedError`` from 

94 ``lexigram.contracts.exceptions.security`` when any guard denies. 

95 """ 

96 

97 def add(self, guard: GuardProtocol) -> GuardChainProtocol: 

98 """Add a guard to the chain. 

99 

100 Args: 

101 guard: The guard to append. 

102 

103 Returns: 

104 Self, for fluent chaining. 

105 """ 

106 ... 

107 

108 async def execute(self, context: dict[str, Any]) -> None: 

109 """Execute all guards in order, raising on first denial. 

110 

111 Args: 

112 context: Arbitrary request context forwarded to each guard. 

113 

114 Raises: 

115 GuardDeniedError: If any guard returns False from ``can_activate``. 

116 """ 

117 ... 

118 

119 

120@runtime_checkable 

121class InputSanitizerProtocol(Protocol): 

122 """Sanitizes raw input strings against injection vectors. 

123 

124 Implementations should strip XSS payloads, HTML entities, and 

125 dangerous character sequences without raising on clean input. 

126 """ 

127 

128 def sanitize(self, value: str) -> str: 

129 """Sanitize a single string value. 

130 

131 Args: 

132 value: Raw input string. 

133 

134 Returns: 

135 The sanitized string. 

136 """ 

137 ... 

138 

139 def sanitize_dict(self, data: dict[str, Any]) -> dict[str, Any]: 

140 """Recursively sanitize all string values in a mapping. 

141 

142 Args: 

143 data: Dictionary whose string leaf values will be sanitized. 

144 

145 Returns: 

146 A new dictionary with all string values sanitized. 

147 """ 

148 ... 

149 

150 def sanitize_header_value(self, value: str) -> str: 

151 """Strip CRLF characters from an HTTP header value to prevent header injection. 

152 

153 Args: 

154 value: Raw header value string. 

155 

156 Returns: 

157 The value with CR and LF characters removed or replaced. 

158 """ 

159 ... 

160 

161 def is_safe_url_for_request(self, url: str) -> bool: 

162 """Return False if the URL targets a private or reserved IP range (SSRF guard). 

163 

164 Args: 

165 url: Fully-qualified URL to evaluate. 

166 

167 Returns: 

168 True if the URL is considered safe to request, False otherwise. 

169 """ 

170 ... 

171 

172 

173@runtime_checkable 

174class SecurityHeadersProtocol(Protocol): 

175 """Applies security-related HTTP response headers. 

176 

177 Implementations must support HSTS, X-Frame-Options, CSP, and 

178 X-Content-Type-Options at minimum. 

179 """ 

180 

181 def apply(self, headers: dict[str, str]) -> dict[str, str]: 

182 """Apply security headers to an existing headers mapping. 

183 

184 Args: 

185 headers: Mutable mapping of header names to values. 

186 

187 Returns: 

188 The headers mapping with security headers merged in. 

189 """ 

190 ... 

191 

192 

193@runtime_checkable 

194class EncryptionProtocol(Protocol): 

195 """Protocol for symmetric/asymmetric encryption providers.""" 

196 

197 async def encrypt(self, plaintext: bytes, key_id: str | None = None) -> bytes: ... 

198 async def decrypt(self, ciphertext: bytes, key_id: str | None = None) -> bytes: ... 

199 async def rotate_key(self, key_id: str) -> str: ... 

200 

201 

202@runtime_checkable 

203class CORSProtocol(Protocol): 

204 """Protocol for Cross-Origin Resource Sharing policy enforcement.""" 

205 

206 def is_allowed_origin(self, origin: str) -> bool: ... 

207 def get_allowed_headers(self) -> list[str]: ... 

208 def get_allowed_methods(self) -> list[str]: ... 

209 def get_max_age(self) -> int: ... 

210 

211 

212@runtime_checkable 

213class CSPProtocol(Protocol): 

214 """Protocol for Content Security Policy header generation.""" 

215 

216 def build_header(self) -> str: ... 

217 def add_directive(self, directive: str, value: str) -> None: ... 

218 

219 

220@runtime_checkable 

221class CSRFProtocol(Protocol): 

222 """Protocol for CSRF token generation and validation.""" 

223 

224 def generate_token(self, session_id: str) -> str: ... 

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

226 def invalidate_token(self, session_id: str) -> None: ... 

227 

228 

229__all__ = [ 

230 "CORSProtocol", 

231 "CSPProtocol", 

232 "CSRFProtocol", 

233 "EncryptionProtocol", 

234 "GuardChainProtocol", 

235 "HasherProtocol", 

236 "InputSanitizerProtocol", 

237 "KeyDerivationProtocol", 

238 "SecurityHeadersProtocol", 

239]