Coverage for app/venv/lib/python3.14/site-packages/weblate/gitexport/views.py: 20%

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

2# 

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

4from __future__ import annotations 

5 

6import os.path 

7import subprocess 

8from base64 import b64decode 

9from contextlib import suppress 

10from email import message_from_string 

11from functools import partial 

12from selectors import EVENT_READ, DefaultSelector 

13from typing import TYPE_CHECKING, BinaryIO, cast 

14 

15from django.core.exceptions import PermissionDenied, SuspiciousOperation 

16from django.http import Http404, StreamingHttpResponse 

17from django.http.response import HttpResponse, HttpResponseServerError 

18from django.shortcuts import redirect 

19from django.urls import reverse 

20from django.views.decorators.cache import never_cache 

21from django.views.decorators.csrf import csrf_exempt 

22 

23from weblate.auth.models import User 

24from weblate.gitexport.utils import find_git_http_backend 

25from weblate.trans.models import Component 

26from weblate.utils.errors import report_error 

27from weblate.utils.views import parse_path 

28from weblate.vcs.models import VCS_REGISTRY 

29 

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

31 from collections.abc import Iterator 

32 

33 from django.http import HttpRequest 

34 from django.http.response import HttpResponseBase 

35 

36 from weblate.auth.models import AuthenticatedHttpRequest 

37 

38 

39def response_authenticate(): 

40 """Return 401 response with authenticate header.""" 

41 response = HttpResponse(status=401) 

42 response["WWW-Authenticate"] = 'Basic realm="Weblate Git access"' 

43 return response 

44 

45 

46def authenticate(request: HttpRequest, auth: str) -> bool: 

47 """Perform authentication with HTTP Basic auth.""" 

48 try: 

49 method, data = auth.split(None, 1) 

50 except (ValueError, TypeError): 

51 return False 

52 if method.lower() == "basic": 

53 try: 

54 username, code = b64decode(data).decode("iso-8859-1").split(":", 1) 

55 except (ValueError, TypeError): 

56 return False 

57 try: 

58 user = User.objects.get(username=username, auth_token__key=code) 

59 except User.DoesNotExist: 

60 return False 

61 

62 if not user.is_active: 

63 return False 

64 

65 request.user = user 

66 return True 

67 return False 

68 

69 

70@never_cache 

71@csrf_exempt 

72def git_export( 

73 request: AuthenticatedHttpRequest, path: list[str], git_request: str 

74) -> HttpResponseBase: 

75 """ 

76 Git HTTP server view. 

77 

78 Wrapper around git-http-backend to provide Git repositories export over HTTP. 

79 Performs permission checks and hands over execution to the wrapper. 

80 """ 

81 # Reject non pull access early 

82 if request.GET.get("service", "") not in {"", "git-upload-pack"}: 

83 msg = "Only pull is supported" 

84 raise PermissionDenied(msg) 

85 

86 # HTTP authentication 

87 auth = request.headers.get("authorization", "") 

88 

89 if auth and not authenticate(request, auth): 

90 return response_authenticate() 

91 

92 try: 

93 obj = parse_path(request, path, (Component,)) 

94 except Http404: 

95 if not request.user.is_authenticated: 

96 return response_authenticate() 

97 raise 

98 # Strip possible double path separators 

99 git_request = git_request.lstrip("/\\") 

100 

101 # Permissions 

102 if not request.user.has_perm("vcs.access", obj): 

103 if not request.user.is_authenticated: 

104 return response_authenticate() 

105 msg = "No VCS permissions" 

106 raise PermissionDenied(msg) 

107 if obj.vcs not in VCS_REGISTRY.git_based: 

108 msg = "Not a git repository" 

109 raise Http404(msg) 

110 if obj.is_repo_link: 

111 return redirect( 

112 "{}?{}".format( 

113 reverse( 

114 "git-export", 

115 kwargs={ 

116 "path": obj.linked_component.get_url_path(), 

117 "git_request": git_request, 

118 }, 

119 ), 

120 request.META["QUERY_STRING"], 

121 ), 

122 permanent=True, 

123 ) 

124 

125 # Invoke Git HTTP backend 

126 wrapper = GitHTTPBackendWrapper(obj, request, git_request) 

127 return wrapper.get_response() 

128 

129 

130class GitStreamingHttpResponse(StreamingHttpResponse): 

131 def __init__(self, streaming_content, *args, **kwargs) -> None: 

132 super().__init__(streaming_content.stream(), *args, **kwargs) 

133 self.wrapper = streaming_content 

134 

135 def close(self) -> None: 

136 if self.wrapper.process.poll() is None: 

137 self.wrapper.process.kill() 

138 self.wrapper.wait() 

139 super().close() 

140 

141 

142class GitHTTPBackendWrapper: 

143 def __init__( 

144 self, obj, request: AuthenticatedHttpRequest, git_request: str 

145 ) -> None: 

146 self.path = os.path.join(obj.full_path, git_request) 

147 self.obj = obj 

148 self.request = request 

149 self.selector = DefaultSelector() 

150 self._headers: bytes = b"" 

151 self._stderr: list[bytes] = [] 

152 self._stdout: list[bytes] = [] 

153 

154 # Find Git HTTP backend 

155 git_http_backend = find_git_http_backend() 

156 if git_http_backend is None: 

157 msg = "git-http-backend not found" 

158 raise SuspiciousOperation(msg) 

159 

160 # Invoke Git HTTP backend 

161 self.process = subprocess.Popen( 

162 [git_http_backend], 

163 env=self.get_env(), 

164 stdin=subprocess.PIPE, 

165 stdout=subprocess.PIPE, 

166 stderr=subprocess.PIPE, 

167 close_fds=True, 

168 bufsize=0, 

169 ) 

170 

171 def get_env(self) -> dict[str, str]: 

172 result = { 

173 "REQUEST_METHOD": self.request.method, 

174 "PATH_TRANSLATED": self.path, 

175 "GIT_HTTP_EXPORT_ALL": "1", 

176 "CONTENT_TYPE": self.request.headers.get("content-type", ""), 

177 "QUERY_STRING": self.request.META.get("QUERY_STRING", ""), 

178 "HTTP_CONTENT_ENCODING": self.request.headers.get("content-encoding", ""), 

179 } 

180 # Fault injection in tests 

181 if "X_WEBLATE_NO_EXPORT" in self.request.headers: 

182 del result["GIT_HTTP_EXPORT_ALL"] 

183 return result 

184 

185 def send_body(self) -> None: 

186 # Fetch request body (it could be streamed) 

187 body = self.request.body 

188 

189 with suppress(BrokenPipeError): 

190 # Ignore broken pipe as that can happen when process failed with an error 

191 self.process.stdin.write(body) # type: ignore[union-attr] 

192 

193 self.process.stdin.close() # type: ignore[union-attr] 

194 

195 def fetch_headers(self) -> None: 

196 """Fetch initial chunk of response to parse headers.""" 

197 while True: 

198 for key, _mask in self.selector.select(timeout=1): 

199 if key.data: 

200 self._stdout.append( 

201 self.process.stdout.read(1024) # type: ignore[union-attr] 

202 ) 

203 headers = b"".join(self._stdout) 

204 if b"\r\n\r\n" in headers: 

205 self._headers, body = headers.split(b"\r\n\r\n", 1) 

206 self._stdout = [body] 

207 else: 

208 self._stderr.append( 

209 self.process.stderr.read() # type: ignore[union-attr] 

210 ) 

211 if self.process.poll() is not None or self._headers is not None: 

212 break 

213 

214 def stream(self) -> Iterator[bytes]: 

215 yield from self._stdout 

216 yield from iter( 

217 partial(self.process.stdout.read, 1024), # type: ignore[union-attr] 

218 b"", 

219 ) 

220 

221 def get_response(self) -> HttpResponseBase: 

222 # Iniciate select() 

223 stdout = cast("BinaryIO", self.process.stdout) 

224 stderr = cast("BinaryIO", self.process.stderr) 

225 self.selector.register(stdout, EVENT_READ, True) 

226 self.selector.register(stderr, EVENT_READ, False) 

227 

228 # Send request body 

229 self.send_body() 

230 

231 # Read initial chunk of response to parse headers 

232 # This assumes that git-http-backend will fail here if it will fail 

233 self.fetch_headers() 

234 

235 self.selector.unregister(stdout) 

236 self.selector.unregister(stderr) 

237 self.selector.close() 

238 

239 retcode = self.process.poll() 

240 

241 output_err = b"".join(self._stderr).decode() 

242 

243 # Log error 

244 if output_err: 

245 report_error( 

246 "Git backend failure", 

247 extra_log=output_err, 

248 project=self.obj.project, 

249 level="error", 

250 message=True, 

251 ) 

252 

253 # Handle failure 

254 if retcode is not None and retcode != 0: 

255 return HttpResponseServerError(output_err) 

256 

257 message = message_from_string(self._headers.decode()) 

258 

259 # Handle status in response 

260 if "status" in message: 

261 self.wait() 

262 return HttpResponse(status=int(message["status"].split()[0])) 

263 

264 # Send streaming content as response 

265 return GitStreamingHttpResponse( 

266 streaming_content=self, content_type=message["content-type"] 

267 ) 

268 

269 def wait(self): 

270 self.process.wait() 

271 if self.process.stdout is not None: 

272 self.process.stdout.close() 

273 if self.process.stderr is not None: 

274 self.process.stderr.close()