Coverage for extra_codeowners/manifest.py: 92%

1"""GitHub App Manifest registration flow.""" 

2 

3from __future__ import annotations 

4 

5import html 

6import json 

7from datetime import UTC, datetime, timedelta 

8from typing import Any 

9from urllib.parse import quote 

10 

11import httpx 

12import jwt 

13 

14from extra_codeowners.settings import Settings 

15 

16 

17class ManifestError(RuntimeError): 

18 """The App Manifest setup handshake failed validation or conversion.""" 

19 

20 

21class ManifestService: 

22 """Build and exchange GitHub App Manifests. 

23 

24 Setup is deployment-controlled and disabled by default. Conversion returns 

25 credentials once, directly to the operator over a no-store response. 

26 """ 

27 

28 def __init__( 

29 self, 

30 settings: Settings, 

31 *, 

32 transport: httpx.AsyncBaseTransport | None = None, 

33 ) -> None: 

34 if not settings.setup_enabled: 

35 msg = "App Manifest setup is disabled" 

36 raise ManifestError(msg) 

37 if settings.public_url is None or settings.setup_state_secret is None: 37 ↛ 38line 37 didn't jump to line 38 because the condition on line 37 was never true

38 msg = "setup requires public_url and setup_state_secret" 

39 raise ManifestError(msg) 

40 self.settings = settings 

41 self._state_secret = settings.setup_state_secret.get_secret_value() 

42 self._http = httpx.AsyncClient( 

43 base_url=str(settings.github_api_url).rstrip("/"), 

44 timeout=20, 

45 headers={ 

46 "Accept": "application/vnd.github+json", 

47 "User-Agent": "extra-codeowners", 

48 "X-GitHub-Api-Version": settings.github_api_version, 

49 }, 

50 transport=transport, 

51 ) 

52 

53 async def close(self) -> None: 

54 """Close pooled HTTP connections.""" 

55 await self._http.aclose() 

56 

57 def issue_state(self) -> str: 

58 """Issue a short-lived state token bound to this setup flow.""" 

59 now = datetime.now(UTC) 

60 return str( 

61 jwt.encode( 

62 { 

63 "iat": int(now.timestamp()), 

64 "exp": int( 

65 (now + timedelta(seconds=self.settings.setup_state_ttl_seconds)).timestamp() 

66 ), 

67 "aud": "extra-codeowners-app-manifest", 

68 }, 

69 self._state_secret, 

70 algorithm="HS256", 

71 ) 

72 ) 

73 

74 def validate_state(self, state: str) -> None: 

75 """Validate setup state signature, expiry, and audience.""" 

76 try: 

77 jwt.decode( 

78 state, 

79 self._state_secret, 

80 algorithms=["HS256"], 

81 audience="extra-codeowners-app-manifest", 

82 options={"require": ["iat", "exp", "aud"]}, 

83 ) 

84 except jwt.PyJWTError as error: 

85 msg = "invalid or expired App Manifest setup state" 

86 raise ManifestError(msg) from error 

87 

88 def manifest(self) -> dict[str, Any]: 

89 """Return the least-privilege GitHub App Manifest.""" 

90 base = str(self.settings.public_url).rstrip("/") 

91 return { 

92 "name": "Extra CODEOWNERS", 

93 "url": "https://github.com/stampbot/extra-codeowners", 

94 "description": ( 

95 "Require human CODEOWNER approval or a narrowly delegated GitHub App approval." 

96 ), 

97 "public": False, 

98 "redirect_url": f"{base}/setup/callback", 

99 "setup_url": f"{base}/setup/complete", 

100 "setup_on_update": True, 

101 "request_oauth_on_install": False, 

102 "hook_attributes": {"url": f"{base}/webhooks/github", "active": True}, 

103 "default_permissions": { 

104 "checks": "write", 

105 "contents": "read", 

106 "members": "read", 

107 "metadata": "read", 

108 "pull_requests": "read", 

109 # GitHub requires this installation permission before an App 

110 # can be selected as an expected source in organization 

111 # rulesets. Runtime tokens are downscoped and cannot write 

112 # commit statuses; this service publishes Check Runs only. 

113 "statuses": "write", 

114 }, 

115 "default_events": [ 

116 "check_run", 

117 "installation_target", 

118 "label", 

119 "member", 

120 "membership", 

121 "organization", 

122 "push", 

123 "pull_request", 

124 "pull_request_review", 

125 "repository", 

126 "team", 

127 "team_add", 

128 ], 

129 } 

130 

131 def registration_page(self, organization: str | None = None) -> str: 

132 """Return an auto-submitting manifest form for user or organization ownership.""" 

133 state = self.issue_state() 

134 if organization: 

135 action_path = f"organizations/{quote(organization, safe='')}/settings/apps/new" 

136 else: 

137 action_path = "settings/apps/new" 

138 action = f"https://github.com/{action_path}?state={quote(state, safe='')}" 

139 manifest_json = json.dumps(self.manifest(), separators=(",", ":")) 

140 return f"""<!doctype html> 

141<html lang="en"> 

142<head><meta charset="utf-8"><meta name="viewport" content="width=device-width"> 

143<title>Create Extra CODEOWNERS GitHub App</title></head> 

144<body> 

145<main> 

146 <h1>Create Extra CODEOWNERS</h1> 

147 <p>Review the requested permissions on GitHub before creating the App.</p> 

148 <form method="post" action="{html.escape(action, quote=True)}"> 

149 <input type="hidden" name="manifest" value="{html.escape(manifest_json, quote=True)}"> 

150 <button type="submit">Continue to GitHub</button> 

151 </form> 

152</main> 

153</body> 

154</html>""" 

155 

156 async def exchange(self, code: str, state: str) -> dict[str, Any]: 

157 """Exchange GitHub's single-use code for newly created App credentials.""" 

158 self.validate_state(state) 

159 response = await self._http.post(f"/app-manifests/{quote(code, safe='')}/conversions") 

160 if not response.is_success: 

161 msg = f"GitHub rejected App Manifest conversion with HTTP {response.status_code}" 

162 raise ManifestError(msg) 

163 value = response.json() 

164 if not isinstance(value, dict): 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true

165 msg = "GitHub returned a malformed App Manifest conversion" 

166 raise ManifestError(msg) 

167 required = ("id", "pem", "webhook_secret") 

168 if any(not value.get(field) for field in required): 

169 msg = "GitHub App Manifest conversion omitted required credentials" 

170 raise ManifestError(msg) 

171 return value 

172 

173 @staticmethod 

174 def credentials_page(credentials: dict[str, Any]) -> str: 

175 """Render the one-time credential result without embedding executable markup.""" 

176 safe_json = html.escape(json.dumps(credentials, indent=2, sort_keys=True)) 

177 return f"""<!doctype html> 

178<html lang="en"> 

179<head><meta charset="utf-8"><meta name="viewport" content="width=device-width"> 

180<title>Store Extra CODEOWNERS credentials</title></head> 

181<body> 

182<main> 

183 <h1>Store the GitHub App credentials now</h1> 

184 <p>This service does not retain this response. Put the values in your secret manager, 

185 then close this page. Never commit them to a repository.</p> 

186 <pre aria-label="New GitHub App credentials">{safe_json}</pre> 

187</main> 

188</body> 

189</html>"""