Coverage for app/venv/lib/python3.14/site-packages/weblate/trans/views/hooks.py: 27%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4from __future__ import annotations 

5 

6import json 

7import re 

8from collections.abc import Callable 

9from typing import TYPE_CHECKING, TypedDict, cast 

10from urllib.parse import quote, urlparse 

11 

12from django.conf import settings 

13from django.db.models import Q 

14from drf_spectacular.utils import OpenApiParameter, extend_schema 

15from rest_framework import parsers, serializers 

16from rest_framework.exceptions import ( 

17 MethodNotAllowed, 

18 NotFound, 

19 ParseError, 

20 ValidationError, 

21) 

22from rest_framework.permissions import AllowAny 

23from rest_framework.request import Request 

24from rest_framework.response import Response 

25from rest_framework.views import APIView 

26 

27from weblate.api.serializers import MultiFieldHyperlinkedIdentityField 

28from weblate.auth.models import User 

29from weblate.logger import LOGGER 

30from weblate.trans.actions import ActionEvents 

31from weblate.trans.models import Component 

32from weblate.trans.tasks import perform_update 

33from weblate.utils.errors import report_error 

34 

35if TYPE_CHECKING: 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true

36 from django.http import QueryDict 

37 

38BITBUCKET_GIT_REPOS = ( 

39 "ssh://git@{server}/{full_name}.git", 

40 "ssh://git@{server}/{full_name}", 

41 "git@{server}:{full_name}.git", 

42 "git@{server}:{full_name}", 

43 "https://{server}/{full_name}.git", 

44 "https://{server}/{full_name}", 

45) 

46 

47BITBUCKET_HG_REPOS = ( 

48 "https://{server}/{full_name}", 

49 "ssh://hg@{server}/{full_name}", 

50 "hg::ssh://hg@{server}/{full_name}", 

51 "hg::https://{server}/{full_name}", 

52) 

53 

54GITHUB_REPOS = ( 

55 "git://github.com/%(owner)s/%(slug)s.git", 

56 "git://github.com/%(owner)s/%(slug)s", 

57 "https://github.com/%(owner)s/%(slug)s.git", 

58 "https://github.com/%(owner)s/%(slug)s", 

59 "git@github.com:%(owner)s/%(slug)s.git", 

60 "git@github.com:%(owner)s/%(slug)s", 

61) 

62 

63PAGURE_REPOS = ( 

64 "https://{server}/{project}", 

65 "https://{server}/{project}.git", 

66 "ssh://git@{server}/{project}", 

67 "ssh://git@{server}/{project}.git", 

68) 

69 

70AZURE_REPOS = ( 

71 "https://dev.azure.com/{organization}/{project}/_git/{repository}", 

72 "https://dev.azure.com/{organization}/{projectId}/_git/{repositoryId}", 

73 "git@ssh.dev.azure.com:v3/{organization}/{project}/{repository}", 

74 "https://{organization}.visualstudio.com/{project}/_git/{repository}", 

75 "{organization}@vs-ssh.visualstudio.com:v3/{organization}/{project}/{repository}", 

76) 

77 

78 

79class HandlerResponse(TypedDict): 

80 service_long_name: str 

81 repo_url: str 

82 repos: list[str] 

83 branch: str | None 

84 full_name: str 

85 

86 

87HandlerType = Callable[[dict, Request | None], HandlerResponse | None] 

88 

89 

90HOOK_HANDLERS: dict[str, HandlerType] = {} 

91 

92 

93def register_hook(handler: HandlerType) -> HandlerType: 

94 """Register hook handler.""" 

95 name = handler.__name__.split("_")[0] 

96 HOOK_HANDLERS[name] = handler 

97 return handler 

98 

99 

100class HookMatchDict(TypedDict): 

101 repository_matches: int 

102 branch_matches: int 

103 enabled_hook_matches: int 

104 

105 

106class HookMatchSerializer(serializers.Serializer): 

107 repository_matches = serializers.IntegerField() 

108 branch_matches = serializers.IntegerField() 

109 enabled_hook_matches = serializers.IntegerField() 

110 

111 

112class HookResponseSerializer(serializers.Serializer): 

113 message = serializers.CharField(max_length=200) 

114 status = serializers.ChoiceField(choices=["success", "failure"]) 

115 match_status = HookMatchSerializer(required=False) 

116 updated_components = MultiFieldHyperlinkedIdentityField( 

117 view_name="api:component-detail", 

118 lookup_field=("project__slug", "slug"), 

119 many=True, 

120 read_only=True, 

121 required=False, 

122 ) 

123 

124 

125class HookRequestSerializer(serializers.Serializer): 

126 """Native webhook payload from the notifying service.""" 

127 

128 

129class PayloadMixin: 

130 @staticmethod 

131 def extract_payload(request_data: QueryDict | parsers.DataAndFiles | dict) -> dict: 

132 data: QueryDict | dict 

133 if isinstance(request_data, parsers.DataAndFiles): 

134 data = request_data.data 

135 else: 

136 data = request_data 

137 try: 

138 payload = cast("str | bytes", data["payload"]) 

139 except KeyError as exc: 

140 msg = "Missing payload parameter!" 

141 raise ParseError(msg) from exc 

142 try: 

143 return json.loads(payload) 

144 except ValueError as exc: 

145 msg = f"JSON parse error - {exc!s}" 

146 raise ParseError(msg) from exc 

147 

148 

149class PayloadMultiPartParser(parsers.MultiPartParser, PayloadMixin): 

150 """Extract payload from application/x-www-form-urlencoded.""" 

151 

152 def parse(self, stream, media_type=None, parser_context=None): 

153 return self.extract_payload( 

154 super().parse(stream, media_type=media_type, parser_context=parser_context) 

155 ) 

156 

157 

158class PayloadFormParserParser(parsers.FormParser, PayloadMixin): 

159 """Extract payload from multipart/form-data.""" 

160 

161 def parse(self, stream, media_type=None, parser_context=None): 

162 return self.extract_payload( 

163 super().parse(stream, media_type=media_type, parser_context=parser_context) 

164 ) 

165 

166 

167@extend_schema( 

168 responses=HookResponseSerializer, 

169 request=HookRequestSerializer, 

170 parameters=[ 

171 OpenApiParameter( 

172 "service", 

173 enum=["bitbucket", "github", "gitea", "gitee", "gitlab", "pagure", "azure"], 

174 location=OpenApiParameter.PATH, 

175 ), 

176 ], 

177) 

178class ServiceHookView(APIView): 

179 authentication_classes = [] # noqa: RUF012 

180 permission_classes = [AllowAny] # noqa: RUF012 

181 http_method_names = ["post"] # noqa: RUF012 

182 parser_classes = ( 

183 parsers.JSONParser, 

184 PayloadMultiPartParser, 

185 PayloadFormParserParser, 

186 ) 

187 

188 def hook_response( 

189 self, 

190 response: str = "Update triggered", 

191 status: int = 200, 

192 match_status: HookMatchDict | None = None, 

193 updated_components: list[Component] | None = None, 

194 ) -> Response: 

195 """Create a hook response.""" 

196 serializer = HookResponseSerializer( 

197 { 

198 "status": "success" if status in {200, 201} else "failure", 

199 "message": response, 

200 "match_status": match_status, 

201 "updated_components": updated_components, 

202 }, 

203 context={"request": self.request}, 

204 ) 

205 return Response(serializer.data, status=status) 

206 

207 def post(self, request: Request, service: str) -> Response: 

208 """Process incoming webhook from a code hosting site.""" 

209 # We support only post methods 

210 if not settings.ENABLE_HOOKS: 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true

211 msg = "POST" 

212 raise MethodNotAllowed(msg) 

213 

214 # Get service helper 

215 try: 

216 hook_helper = HOOK_HANDLERS[service] 

217 except KeyError as exc: 

218 msg = f"Hook {service} not supported" 

219 raise NotFound(msg) from exc 

220 

221 # Check if we got payload 

222 data = request.data 

223 if not data: 223 ↛ 228line 223 didn't jump to line 228 because the condition on line 223 was always true

224 msg = "Invalid data in json payload!" 

225 raise ValidationError(msg) 

226 

227 # Send the request data to the service handler. 

228 try: 

229 service_data = hook_helper(data, request) 

230 except Exception as exc: 

231 LOGGER.error("failed to parse service %s data", service) 

232 report_error("Invalid service data") 

233 msg = "Invalid data in json payload!" 

234 raise ValidationError(msg) from exc 

235 

236 # This happens on ping request upon installation 

237 if service_data is None: 

238 return self.hook_response("Hook working", status=201) 

239 

240 # Log data 

241 service_long_name = service_data["service_long_name"] 

242 repos = service_data["repos"] 

243 repo_url = service_data["repo_url"] 

244 branch = service_data["branch"] 

245 full_name = service_data["full_name"] 

246 

247 # Generate filter 

248 spfilter = ( 

249 Q(repo__in=repos) 

250 | Q(repo__iendswith=full_name) 

251 | Q(repo__iendswith=f"{full_name}/") 

252 | Q(repo__iendswith=f"{full_name}.git") 

253 ) 

254 user = User.objects.get_or_create_bot( 

255 scope="webhook", 

256 name=service, 

257 verbose=f"{service_data['service_long_name']} webhook", 

258 ) 

259 

260 for repo in repos: 

261 # We need to match also URLs which include username and password 

262 if repo.startswith("http://"): 

263 spfilter |= Q(repo__startswith="http://") & Q( 

264 repo__endswith=f"@{repo[7:]}" 

265 ) 

266 elif repo.startswith("https://"): 

267 spfilter |= Q(repo__startswith="https://") & Q( 

268 repo__endswith=f"@{repo[8:]}" 

269 ) 

270 # Include URLs with trailing slash 

271 spfilter |= Q(repo=repo + "/") 

272 

273 all_components = repo_components = Component.objects.filter(spfilter) 

274 

275 if branch is not None: 

276 all_components = repo_components.filter(branch=branch) 

277 

278 all_components_count = all_components.count() 

279 repo_components_count = repo_components.count() 

280 enabled_components = all_components.filter(project__enable_hooks=True) 

281 

282 LOGGER.info( 

283 "received %s notification on repository %s, URL %s, branch %s, " 

284 "%d matching components, %d to process, %d linked", 

285 service_long_name, 

286 full_name, 

287 repo_url, 

288 branch, 

289 all_components_count, 

290 len(enabled_components), 

291 Component.objects.filter(linked_component__in=enabled_components).count(), 

292 ) 

293 

294 # Trigger updates 

295 updates = 0 

296 for obj in enabled_components: 

297 updates += 1 

298 LOGGER.info("%s notification will update %s", service_long_name, obj) 

299 obj.change_set.create( 

300 action=ActionEvents.HOOK, details=service_data, user=user 

301 ) 

302 perform_update.delay("Component", obj.pk, user_id=user.id) 

303 

304 match_status = HookMatchDict( 

305 repository_matches=repo_components_count, 

306 branch_matches=all_components_count, 

307 enabled_hook_matches=len(enabled_components), 

308 ) 

309 

310 if updates == 0: 

311 return self.hook_response( 

312 "No matching repositories found!", 

313 status=202, 

314 match_status=match_status, 

315 ) 

316 

317 updated_components = [obj.full_slug for obj in enabled_components] 

318 

319 return self.hook_response( 

320 "Update triggered: {}".format(", ".join(updated_components)), 

321 match_status=match_status, 

322 updated_components=updated_components, 

323 ) 

324 

325 

326def bitbucket_extract_changes(data: dict) -> list[dict]: 

327 if "changes" in data: 

328 return data["changes"] 

329 if "push" in data: 

330 return data["push"]["changes"] 

331 if "commits" in data: 

332 return data["commits"] 

333 return [] 

334 

335 

336def bitbucket_extract_branch(data: dict) -> str | None: 

337 changes = bitbucket_extract_changes(data) 

338 if changes: 

339 last = changes[-1] 

340 if "branch" in last: 

341 return last["branch"] 

342 if last.get("new"): 

343 return changes[-1]["new"]["name"] 

344 if last.get("old"): 

345 return changes[-1]["old"]["name"] 

346 if "ref" in last: 

347 return last["ref"]["displayId"] 

348 # Pullrequest merged action 

349 if "pullrequest" in data: 

350 return data["pullrequest"]["destination"]["branch"]["name"] 

351 return None 

352 

353 

354def bitbucket_extract_full_name(repository: dict) -> str: 

355 if "full_name" in repository: 

356 return repository["full_name"] 

357 if "fullName" in repository: 

358 return repository["fullName"] 

359 if "owner" in repository and "slug" in repository: 

360 return "{}/{}".format(repository["owner"], repository["slug"]) 

361 if "project" in repository and "slug" in repository: 

362 return "{}/{}".format(repository["project"]["key"], repository["slug"]) 

363 msg = "Could not determine repository full name" 

364 raise ValueError(msg) 

365 

366 

367def bitbucket_extract_repo_url(data, repository: dict) -> str: 

368 if "links" in repository: 

369 if "html" in repository["links"]: 

370 return repository["links"]["html"]["href"] 

371 return repository["links"]["self"][0]["href"] 

372 if "canon_url" in data: 

373 return "{}{}".format(data["canon_url"], repository["absolute_url"]) 

374 msg = "Could not determine repository URL" 

375 raise ValueError(msg) 

376 

377 

378@register_hook 

379def bitbucket_hook_helper(data, request: Request | None) -> HandlerResponse | None: 

380 """Parse service hook from Bitbucket.""" 

381 # Bitbucket ping event 

382 if request and request.headers.get("x-event-key") not in { 

383 "repo:push", 

384 "repo:refs_changed", 

385 "pullrequest:fulfilled", 

386 "pr:merged", 

387 }: 

388 return None 

389 

390 if "pullRequest" in data: 

391 # The pr:merged event 

392 repository = data["pullRequest"]["fromRef"]["repository"] 

393 else: 

394 repository = data["repository"] 

395 full_name = bitbucket_extract_full_name(repository) 

396 repo_url = bitbucket_extract_repo_url(data, repository) 

397 

398 # Extract repository links 

399 if "links" in repository and "clone" in repository["links"]: 

400 repos = [val["href"] for val in repository["links"]["clone"]] 

401 else: 

402 repo_servers = {"bitbucket.org", urlparse(repo_url).hostname} 

403 repos = [] 

404 templates: tuple[str, ...] 

405 if "scm" not in data["repository"]: 

406 templates = BITBUCKET_GIT_REPOS + BITBUCKET_HG_REPOS 

407 elif data["repository"]["scm"] == "hg": 

408 templates = BITBUCKET_HG_REPOS 

409 else: 

410 templates = BITBUCKET_GIT_REPOS 

411 # Construct possible repository URLs 

412 for repo in templates: 

413 repos.extend( 

414 repo.format(full_name=full_name, server=server) 

415 for server in repo_servers 

416 ) 

417 

418 if not repos: 

419 LOGGER.error("unsupported repository: %s", repr(data["repository"])) 

420 msg = "unsupported repository" 

421 raise ValueError(msg) 

422 

423 return { 

424 "service_long_name": "Bitbucket", 

425 "repo_url": repo_url, 

426 "repos": repos, 

427 "branch": bitbucket_extract_branch(data), 

428 "full_name": full_name, 

429 } 

430 

431 

432@register_hook 

433def github_hook_helper(data: dict, request: Request | None) -> HandlerResponse | None: 

434 """Parse hooks from GitHub.""" 

435 # Ignore non push events 

436 if request and request.headers.get("x-github-event") != "push": 

437 return None 

438 # Parse owner, branch and repository name 

439 o_data = data["repository"]["owner"] 

440 owner = o_data["login"] if "login" in o_data else o_data["name"] 

441 slug = data["repository"]["name"] 

442 branch = re.sub(r"^refs/heads/", "", data["ref"]) 

443 

444 params = {"owner": owner, "slug": slug} 

445 

446 if "clone_url" not in data["repository"]: 

447 # Construct possible repository URLs 

448 repos = [repo % params for repo in GITHUB_REPOS] 

449 else: 

450 repos = [] 

451 keys = ["clone_url", "git_url", "ssh_url", "svn_url", "html_url", "url"] 

452 for key in keys: 

453 url = data["repository"].get(key) 

454 if not url: 

455 continue 

456 repos.append(url) 

457 if url.endswith(".git"): 

458 repos.append(url[:-4]) 

459 

460 return { 

461 "service_long_name": "GitHub", 

462 "repo_url": data["repository"]["url"], 

463 "repos": sorted(set(repos)), 

464 "branch": branch, 

465 "full_name": f"{owner}/{slug}", 

466 } 

467 

468 

469@register_hook 

470def gitea_hook_helper(data: dict, request: Request | None) -> HandlerResponse | None: 

471 return { 

472 "service_long_name": "Gitea", 

473 "repo_url": data["repository"]["html_url"], 

474 "repos": [ 

475 data["repository"]["clone_url"], 

476 data["repository"]["ssh_url"], 

477 data["repository"]["html_url"], 

478 ], 

479 "branch": re.sub(r"^refs/heads/", "", data["ref"]), 

480 "full_name": data["repository"]["full_name"], 

481 } 

482 

483 

484@register_hook 

485def gitee_hook_helper(data: dict, request: Request | None) -> HandlerResponse | None: 

486 return { 

487 "service_long_name": "Gitee", 

488 "repo_url": data["repository"]["html_url"], 

489 "repos": [ 

490 data["repository"]["git_http_url"], 

491 data["repository"]["git_ssh_url"], 

492 data["repository"]["git_url"], 

493 data["repository"]["ssh_url"], 

494 data["repository"]["html_url"], 

495 ], 

496 "branch": re.sub(r"^refs/heads/", "", data["ref"]), 

497 "full_name": data["repository"]["path_with_namespace"], 

498 } 

499 

500 

501@register_hook 

502def gitlab_hook_helper(data: dict, request: Request | None) -> HandlerResponse | None: 

503 """Parse hook from GitLab.""" 

504 # Ignore non known events 

505 if "ref" not in data: 

506 return None 

507 ssh_url = data["repository"]["url"] 

508 http_url = ".".join((data["repository"]["homepage"], "git")) 

509 branch = re.sub(r"^refs/heads/", "", data["ref"]) 

510 

511 # Construct possible repository URLs 

512 repos = [ 

513 ssh_url, 

514 http_url, 

515 data["repository"]["git_http_url"], 

516 data["repository"]["git_ssh_url"], 

517 data["repository"]["homepage"], 

518 ] 

519 

520 return { 

521 "service_long_name": "GitLab", 

522 "repo_url": data["repository"]["homepage"], 

523 "repos": repos, 

524 "branch": branch, 

525 "full_name": data["project"]["path_with_namespace"], 

526 } 

527 

528 

529@register_hook 

530def pagure_hook_helper(data: dict, request: Request | None) -> HandlerResponse | None: 

531 """Parse hook from Pagure.""" 

532 # Ignore non known events 

533 if "msg" not in data or data.get("topic") != "git.receive": 

534 return None 

535 

536 server = urlparse(data["msg"]["pagure_instance"]).hostname 

537 project = data["msg"]["project_fullname"] 

538 

539 repos = [repo.format(server=server, project=project) for repo in PAGURE_REPOS] 

540 

541 return { 

542 "service_long_name": "Pagure", 

543 "repo_url": repos[0], 

544 "repos": repos, 

545 "branch": data["msg"]["branch"], 

546 "full_name": project, 

547 } 

548 

549 

550def expand_quoted(name: str): 

551 yield name 

552 quoted = quote(name) 

553 if quoted != name: 

554 yield quoted 

555 

556 

557@register_hook 

558def azure_hook_helper(data: dict, request: Request | None) -> HandlerResponse | None: 

559 if data.get("eventType") != "git.push": 

560 return None 

561 

562 http_url = data["resource"]["repository"]["remoteUrl"] 

563 branch = re.sub(r"^refs/heads/", "", data["resource"]["refUpdates"][0]["name"]) 

564 project = data["resource"]["repository"]["project"]["name"] 

565 projectid = data["resource"]["repository"]["project"]["id"] 

566 repository = data["resource"]["repository"]["name"] 

567 repositoryid = data["resource"]["repository"]["id"] 

568 

569 match = re.match( 

570 r"^https?:\/\/dev\.azure\.com\/" 

571 r"(?P<organization>[a-zA-Z0-9]+[a-zA-Z0-9-]*[a-zA-Z0-9]*)", 

572 http_url, 

573 ) 

574 

575 # Fallback to support old url structure {organization}.visualstudio.com 

576 if match is None: 

577 match = re.match( 

578 r"^https?:\/\/" 

579 r"(?P<organization>[a-zA-Z0-9]+[a-zA-Z0-9-]*[a-zA-Z0-9]*)" 

580 r"\.visualstudio\.com", 

581 http_url, 

582 ) 

583 organization = None 

584 

585 if match is not None: 

586 organization = match.group("organization") 

587 

588 if organization is not None: 

589 repos = [ 

590 repo.format( 

591 organization=organization, 

592 project=e_project, 

593 projectId=projectid, 

594 repository=e_repository, 

595 repositoryId=repositoryid, 

596 ) 

597 for repo in AZURE_REPOS 

598 for e_project in expand_quoted(project) 

599 for e_repository in expand_quoted(repository) 

600 ] 

601 else: 

602 repos = [http_url] 

603 

604 return { 

605 "service_long_name": "Azure", 

606 "repo_url": http_url, 

607 "repos": repos, 

608 "branch": branch, 

609 "full_name": repository, 

610 }