Coverage for extra_codeowners/settings.py: 85%

1"""Process-level configuration loaded from environment variables.""" 

2 

3from __future__ import annotations 

4 

5from functools import cached_property 

6from pathlib import Path 

7from typing import Literal, Self 

8 

9from pydantic import AnyHttpUrl, Field, SecretStr, field_validator, model_validator 

10from pydantic_settings import BaseSettings, SettingsConfigDict 

11from sqlalchemy.engine import make_url 

12from sqlalchemy.exc import ArgumentError 

13 

14 

15class Settings(BaseSettings): 

16 """Runtime settings. 

17 

18 Repository policy can narrow delegated authority but cannot alter these 

19 deployment-level trust settings. 

20 """ 

21 

22 model_config = SettingsConfigDict( 

23 env_file=".env", 

24 env_file_encoding="utf-8", 

25 env_prefix="EXTRA_CODEOWNERS_", 

26 extra="forbid", 

27 validate_default=True, 

28 ) 

29 

30 environment: Literal["development", "test", "production"] = "development" 

31 log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO" 

32 host: str = "127.0.0.1" 

33 port: int = Field(default=8000, ge=1, le=65535) 

34 

35 github_app_id: int | None = Field(default=None, gt=0) 

36 github_private_key: SecretStr | None = None 

37 github_private_key_file: Path | None = None 

38 github_webhook_secret: SecretStr | None = None 

39 github_webhook_secret_file: Path | None = None 

40 github_api_url: AnyHttpUrl = AnyHttpUrl("https://api.github.com") 

41 github_api_version: str = "2026-03-10" 

42 public_url: AnyHttpUrl | None = None 

43 

44 database_url: SecretStr = SecretStr("sqlite:///./extra-codeowners.db") 

45 worker_enabled: bool = True 

46 worker_poll_seconds: float = Field(default=0.5, ge=0.05, le=60) 

47 worker_lease_seconds: int = Field(default=120, ge=30, le=3600) 

48 worker_retry_max_seconds: int = Field(default=60, ge=5, le=3600) 

49 webhook_invalidation_timeout_seconds: float = Field(default=5.0, ge=0.1, le=8.0) 

50 reconcile_enabled: bool = True 

51 reconcile_interval_seconds: int = Field(default=300, ge=60, le=86400) 

52 webhook_delivery_retention_days: int = Field(default=30, ge=1, le=3650) 

53 

54 org_config_repository: str = ".github" 

55 policy_path: str = ".github/extra-codeowners.toml" 

56 check_name: str = "Extra CODEOWNERS / approval" 

57 allow_insecure_changes: bool = False 

58 

59 setup_enabled: bool = False 

60 setup_state_secret: SecretStr | None = None 

61 setup_state_ttl_seconds: int = Field(default=600, ge=60, le=3600) 

62 

63 @field_validator("org_config_repository") 

64 @classmethod 

65 def validate_org_config_repository(cls, value: str) -> str: 

66 name = value.strip() 

67 if not name or len(name) > 100 or name in {".", ".."} or "/" in name or "\\" in name: 67 ↛ 68line 67 didn't jump to line 68 because the condition on line 67 was never true

68 raise ValueError("org_config_repository must be a literal repository name") 

69 return name.lower() 

70 

71 @field_validator("policy_path") 

72 @classmethod 

73 def validate_policy_path(cls, value: str) -> str: 

74 path = value.strip() 

75 if ( 75 ↛ 82line 75 didn't jump to line 82 because the condition on line 75 was never true

76 not path 

77 or path.startswith("/") 

78 or "\\" in path 

79 or any(part in {"", ".", ".."} for part in path.split("/")) 

80 or any(character in path for character in "*?[]!") 

81 ): 

82 raise ValueError("policy_path must be a literal relative POSIX path") 

83 return path 

84 

85 @field_validator("check_name") 

86 @classmethod 

87 def validate_check_name(cls, value: str) -> str: 

88 name = value.strip() 

89 if not name or len(name) > 255 or not name.isprintable(): 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true

90 raise ValueError("check_name must be a printable value containing 1..255 characters") 

91 return name 

92 

93 @model_validator(mode="after") 

94 def validate_secret_sources(self) -> Self: 

95 """Reject ambiguous secret sources and incomplete setup mode.""" 

96 pairs = ( 

97 ("github_private_key", self.github_private_key, self.github_private_key_file), 

98 ( 

99 "github_webhook_secret", 

100 self.github_webhook_secret, 

101 self.github_webhook_secret_file, 

102 ), 

103 ) 

104 for name, inline, file_path in pairs: 

105 if inline is not None and file_path is not None: 

106 msg = f"set only one of {name} and {name}_file" 

107 raise ValueError(msg) 

108 if self.setup_enabled: 

109 if self.setup_state_secret is None: 

110 msg = "setup_state_secret is required when setup_enabled is true" 

111 raise ValueError(msg) 

112 if len(self.setup_state_secret.get_secret_value().encode()) < 32: 

113 msg = "setup_state_secret must contain at least 32 bytes" 

114 raise ValueError(msg) 

115 if self.public_url is None: 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true

116 msg = "public_url is required when setup_enabled is true" 

117 raise ValueError(msg) 

118 if self.public_url.scheme != "https": 

119 msg = "public_url must use HTTPS when setup_enabled is true" 

120 raise ValueError(msg) 

121 if ( 121 ↛ 128line 121 didn't jump to line 128 because the condition on line 121 was never true

122 self.public_url.username is not None 

123 or self.public_url.password is not None 

124 or self.public_url.query is not None 

125 or self.public_url.fragment is not None 

126 or self.public_url.path not in {None, "", "/"} 

127 ): 

128 msg = "public_url must be an origin without credentials, path, query, or fragment" 

129 raise ValueError(msg) 

130 return self 

131 

132 @staticmethod 

133 def _read_secret( 

134 inline: SecretStr | None, 

135 file_path: Path | None, 

136 *, 

137 expand_newlines: bool = False, 

138 ) -> str | None: 

139 if inline is not None: 

140 value = inline.get_secret_value() 

141 elif file_path is not None: 141 ↛ 151line 141 didn't jump to line 151 because the condition on line 141 was always true

142 value = file_path.read_text(encoding="utf-8") 

143 # Kubernetes and Docker secret files conventionally contain one 

144 # terminal line ending. Preserve every other byte, including 

145 # intentional spaces in webhook secrets. 

146 if value.endswith("\r\n"): 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true

147 value = value[:-2] 

148 elif value.endswith("\n"): 148 ↛ 152line 148 didn't jump to line 152 because the condition on line 148 was always true

149 value = value[:-1] 

150 else: 

151 return None 

152 return value.replace("\\n", "\n") if expand_newlines else value 

153 

154 @cached_property 

155 def private_key_value(self) -> str | None: 

156 """Return the App private key without exposing it in model output.""" 

157 return self._read_secret( 

158 self.github_private_key, 

159 self.github_private_key_file, 

160 expand_newlines=True, 

161 ) 

162 

163 @cached_property 

164 def webhook_secret_value(self) -> str | None: 

165 """Return the webhook secret without exposing it in model output.""" 

166 return self._read_secret(self.github_webhook_secret, self.github_webhook_secret_file) 

167 

168 @property 

169 def github_ready(self) -> bool: 

170 """Whether required credentials for webhook processing are present.""" 

171 return ( 

172 self.github_app_id is not None 

173 and bool(self.private_key_value) 

174 and bool(self.webhook_secret_value) 

175 ) 

176 

177 def validate_for_service(self) -> None: 

178 """Fail startup when a production service lacks GitHub credentials.""" 

179 if self.environment == "production" and not self.github_ready: 

180 msg = "production requires GitHub App ID, private key, and webhook secret" 

181 raise ValueError(msg) 

182 if self.environment == "production": 

183 assert self.webhook_secret_value is not None 

184 if len(self.webhook_secret_value.encode()) < 32: 184 ↛ 185line 184 didn't jump to line 185 because the condition on line 184 was never true

185 msg = "production requires a webhook secret containing at least 32 bytes" 

186 raise ValueError(msg) 

187 if self.github_api_url.scheme != "https": 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true

188 msg = "production requires an HTTPS GitHub API URL" 

189 raise ValueError(msg) 

190 try: 

191 database_url = make_url(self.database_url.get_secret_value()) 

192 except ArgumentError as error: 

193 msg = "production requires a valid PostgreSQL database URL" 

194 raise ValueError(msg) from error 

195 if database_url.get_backend_name() != "postgresql": 

196 msg = "production requires a PostgreSQL database URL" 

197 raise ValueError(msg) 

198 ssl_mode = database_url.query.get("sslmode") 

199 query_host = database_url.query.get("host") 

200 effective_host = query_host if query_host is not None else database_url.host 

201 has_routing_override = ( 

202 "hostaddr" in database_url.query or "service" in database_url.query 

203 ) 

204 local_transport = not has_routing_override and ( 

205 ( 

206 isinstance(effective_host, str) 

207 and ( 

208 effective_host in {"localhost", "127.0.0.1", "::1"} 

209 or effective_host.startswith("/") 

210 ) 

211 ) 

212 or effective_host is None 

213 ) 

214 if ssl_mode != "verify-full" and not local_transport: 

215 msg = ( 

216 "remote production PostgreSQL must use sslmode=verify-full; a local " 

217 "socket/proxy transport may omit TLS" 

218 ) 

219 raise ValueError(msg) 

220 

221 def is_organization_config_repository(self, repository_full_name: str) -> bool: 

222 """Return whether a repository is this owner's shared policy source.""" 

223 _, separator, repository_name = repository_full_name.partition("/") 

224 return bool(separator) and repository_name.lower() == self.org_config_repository.lower() 

225 

226 

227def get_settings() -> Settings: 

228 """Construct settings from the current process environment.""" 

229 return Settings()