Coverage for agentos/tools/cors.py: 0%

1""" 

2CORS — Cross-Origin Resource Sharing middleware helper. 

3 

4Supports: 

5 - Fluent builder for CORS configuration 

6 - Origin validation (allowlist/regex) 

7 - Method (GET/POST/PUT/DELETE/etc.) and header control 

8 - Preflight (OPTIONS) response builder 

9 - Max-Age, Credentials, Expose-Headers 

10 - Serialize to dict of response headers 

11""" 

12 

13from __future__ import annotations 

14 

15import re 

16from re import Pattern 

17 

18# ============================================================================ 

19# CORS 

20# ============================================================================ 

21 

22SAFELISTED_METHODS = frozenset({"GET", "HEAD", "POST"}) 

23SAFELISTED_HEADERS = frozenset( 

24 { 

25 "accept", 

26 "accept-language", 

27 "content-language", 

28 "content-type", 

29 } 

30) 

31 

32 

33class CORSConfig: 

34 """CORS configuration builder. 

35 

36 Usage: 

37 cors = (CORSConfig() 

38 .allow_origins("https://example.com", "https://app.example.com") 

39 .allow_methods("GET", "POST", "PUT") 

40 .allow_headers("Content-Type", "Authorization") 

41 .allow_credentials() 

42 .max_age(3600) 

43 ) 

44 

45 # Check if origin is allowed 

46 ok = cors.is_origin_allowed("https://example.com") 

47 

48 # Build preflight response headers 

49 headers = cors.preflight_headers("https://example.com") 

50 """ 

51 

52 def __init__(self): 

53 self._origins: list[str] = [] 

54 self._origin_patterns: list[Pattern] = [] 

55 self._allow_any_origin: bool = False 

56 self._methods: set[str] = set() 

57 self._headers: set[str] = set() 

58 self._expose_headers: set[str] = set() 

59 self._allow_credentials: bool = False 

60 self._max_age: int | None = None 

61 

62 # ---------- Fluent setters ---------- 

63 

64 def allow_origins(self, *origins: str) -> CORSConfig: 

65 for origin in origins: 

66 if origin == "*": 

67 self._allow_any_origin = True 

68 elif "*" in origin and origin != "*": 

69 # Convert glob to regex 

70 pattern = re.escape(origin).replace(r"\*", ".*") 

71 self._origin_patterns.append(re.compile(f"^{pattern}$")) 

72 else: 

73 self._origins.append(origin.rstrip("/")) 

74 return self 

75 

76 def allow_methods(self, *methods: str) -> CORSConfig: 

77 self._methods.update(m.upper() for m in methods) 

78 return self 

79 

80 def allow_headers(self, *headers: str) -> CORSConfig: 

81 self._headers.update(h.lower() for h in headers) 

82 return self 

83 

84 def expose_headers(self, *headers: str) -> CORSConfig: 

85 self._expose_headers.update(h.lower() for h in headers) 

86 return self 

87 

88 def allow_credentials(self) -> CORSConfig: 

89 self._allow_credentials = True 

90 return self 

91 

92 def max_age(self, seconds: int) -> CORSConfig: 

93 self._max_age = seconds 

94 return self 

95 

96 # ---------- Convenience ---------- 

97 

98 def allow_all_origins(self) -> CORSConfig: 

99 self._allow_any_origin = True 

100 return self 

101 

102 def allow_all_methods(self) -> CORSConfig: 

103 self._methods = {"GET", "HEAD", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"} 

104 return self 

105 

106 def allow_all_headers(self) -> CORSConfig: 

107 self._headers = {"*"} 

108 return self 

109 

110 # ---------- Logic ---------- 

111 

112 def is_origin_allowed(self, origin: str) -> bool: 

113 if self._allow_any_origin: 

114 return True 

115 origin = origin.rstrip("/") 

116 if origin in self._origins: 

117 return True 

118 for pattern in self._origin_patterns: 

119 if pattern.match(origin): 

120 return True 

121 return False 

122 

123 def preflight_headers( 

124 self, 

125 origin: str, 

126 request_method: str | None = None, 

127 request_headers: list[str] | None = None, 

128 ) -> dict[str, str]: 

129 """Build response headers for a preflight OPTIONS request.""" 

130 headers: dict = {} 

131 

132 if not self.is_origin_allowed(origin): 

133 return headers 

134 

135 if self._allow_any_origin and not self._allow_credentials: 

136 headers["Access-Control-Allow-Origin"] = "*" 

137 else: 

138 headers["Access-Control-Allow-Origin"] = origin 

139 

140 if self._allow_credentials: 

141 headers["Access-Control-Allow-Credentials"] = "true" 

142 

143 # Methods — return all allowed methods (preflight spec) 

144 if self._methods: 

145 headers["Access-Control-Allow-Methods"] = ", ".join(sorted(self._methods)) 

146 elif request_method: 

147 headers["Access-Control-Allow-Methods"] = request_method.upper() 

148 

149 # Headers 

150 if request_headers: 

151 allowed = set(h.lower() for h in request_headers) 

152 if self._headers and "*" not in self._headers: 

153 allowed &= self._headers 

154 if allowed: 

155 headers["Access-Control-Allow-Headers"] = ", ".join(sorted(allowed)) 

156 elif self._headers: 

157 headers["Access-Control-Allow-Headers"] = ", ".join(sorted(self._headers)) 

158 

159 # Expose 

160 if self._expose_headers: 

161 headers["Access-Control-Expose-Headers"] = ", ".join(sorted(self._expose_headers)) 

162 

163 # Max-Age 

164 if self._max_age is not None: 

165 headers["Access-Control-Max-Age"] = str(self._max_age) 

166 

167 return headers 

168 

169 def actual_headers(self, origin: str) -> dict[str, str]: 

170 """Build response headers for the actual (non-preflight) request.""" 

171 headers: dict = {} 

172 

173 if not self.is_origin_allowed(origin): 

174 return headers 

175 

176 if self._allow_any_origin and not self._allow_credentials: 

177 headers["Access-Control-Allow-Origin"] = "*" 

178 else: 

179 headers["Access-Control-Allow-Origin"] = origin 

180 

181 if self._allow_credentials: 

182 headers["Access-Control-Allow-Credentials"] = "true" 

183 

184 if self._expose_headers: 

185 headers["Access-Control-Expose-Headers"] = ", ".join(sorted(self._expose_headers)) 

186 

187 return headers 

188 

189 def is_preflight(self, method: str, headers: list[str] | None = None) -> bool: 

190 """Check if a request is a CORS preflight request.""" 

191 if method.upper() != "OPTIONS": 

192 return False 

193 # Preflight requires Origin header + either non-safelisted method or custom header 

194 # (Origin header presence is assumed by the caller) 

195 return True