# web.py -- WSGI smart-http server # Copyright (C) 2010 Google, Inc. # Copyright (C) 2012 Jelmer Vernooij # # SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as published by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """HTTP server for dulwich that implements the git smart HTTP protocol.""" __all__ = [ "HTTP_FORBIDDEN", "HTTP_NOT_FOUND", "HTTP_OK", "MAX_CHUNK_SIZE", "MAX_REQUEST_SIZE", "ChunkedEncodingError", "GunzipFilter", "HTTPGitApplication", "HTTPGitRequest", "LimitedInputFilter", "WSGIRequestHandlerLogger", "WSGIServerLogger", "date_time_string", "generate_info_refs", "generate_objects_info_packs", "get_info_packs", "get_info_refs", "get_loose_object", "get_pack_file", "get_text_file", "handle_service_request", "main", "make_server", "make_wsgi_chain", "send_file", ] import os import re import sys import time from collections.abc import Callable, Iterable, Iterator, Sequence from io import BytesIO from types import TracebackType from typing import ( TYPE_CHECKING, Any, BinaryIO, ClassVar, cast, ) from urllib.parse import parse_qs from wsgiref.simple_server import ( ServerHandler, WSGIRequestHandler, WSGIServer, make_server, ) # wsgiref.types was added in Python 3.11 if sys.version_info >= (3, 11): from wsgiref.types import StartResponse, WSGIApplication, WSGIEnvironment else: # Fallback type definitions for Python < 3.11 if TYPE_CHECKING: # For type checking, use the _typeshed types if available try: from _typeshed.wsgi import StartResponse, WSGIApplication, WSGIEnvironment except ImportError: # Define our own protocol types for type checking from typing import Protocol as TypingProtocol class StartResponse(TypingProtocol): # type: ignore[no-redef] """WSGI start_response callable protocol.""" def __call__( self, status: str, response_headers: list[tuple[str, str]], exc_info: tuple[type, BaseException, TracebackType] | None = None, ) -> Callable[[bytes], None]: """Start the response with status and headers.""" ... WSGIEnvironment = dict[str, Any] # type: ignore[misc] WSGIApplication = Callable[ # type: ignore[misc] [WSGIEnvironment, StartResponse], Iterable[bytes] ] else: # At runtime, just use type aliases since these are only for type hints StartResponse = Any WSGIEnvironment = dict[str, Any] WSGIApplication = Callable from dulwich import log_utils from .errors import NotGitRepository from .objects import ObjectID from .protocol import ReceivableProtocol from .repo import BaseRepo, Repo from .server import ( DEFAULT_HANDLERS, Backend, DictBackend, Handler, generate_info_refs, generate_objects_info_packs, ) if TYPE_CHECKING: from typing import Protocol as TypingProtocol from .protocol import Protocol class HandlerConstructor(TypingProtocol): """Protocol for handler constructors.""" def __call__( self, backend: Backend, args: list[bytes], proto: Protocol, stateless_rpc: bool = False, advertise_refs: bool = False, ) -> Handler: """Create a handler instance. Args: backend: The backend to use for the handler args: Arguments for the handler proto: Protocol object for communication stateless_rpc: Whether to use stateless RPC mode advertise_refs: Whether to advertise references Returns: A Handler instance """ ... logger = log_utils.getLogger(__name__) # HTTP error strings HTTP_OK = "200 OK" HTTP_NOT_FOUND = "404 Not Found" HTTP_FORBIDDEN = "403 Forbidden" HTTP_ERROR = "500 Internal Server Error" NO_CACHE_HEADERS = [ ("Expires", "Fri, 01 Jan 1980 00:00:00 GMT"), ("Pragma", "no-cache"), ("Cache-Control", "no-cache, max-age=0, must-revalidate"), ] def cache_forever_headers(now: float | None = None) -> list[tuple[str, str]]: """Generate headers for caching forever. Args: now: Timestamp to use as base (defaults to current time) Returns: List of (header_name, header_value) tuples for caching forever """ if now is None: now = time.time() return [ ("Date", date_time_string(now)), ("Expires", date_time_string(now + 31536000)), ("Cache-Control", "public, max-age=31536000"), ] def date_time_string(timestamp: float | None = None) -> str: """Convert a timestamp to an HTTP date string. Args: timestamp: Unix timestamp to convert (defaults to current time) Returns: HTTP date string in RFC 1123 format """ # From BaseHTTPRequestHandler.date_time_string in BaseHTTPServer.py in the # Python 2.6.5 standard library, following modifications: # - Made a global rather than an instance method. # - weekdayname and monthname are renamed and locals rather than class # variables. # Copyright (c) 2001-2010 Python Software Foundation; All Rights Reserved weekdays = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"] months = [ None, "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec", ] if timestamp is None: timestamp = time.time() year, month, day, hh, mm, ss, wd = time.gmtime(timestamp)[:7] return "%s, %02d %3s %4d %02d:%02d:%02d GMD" % ( # noqa: UP031 weekdays[wd], day, months[month], year, hh, mm, ss, ) def url_prefix(mat: re.Match[str]) -> str: """Extract the URL prefix from a regex match. Args: mat: A regex match object. Returns: The URL prefix, defined as the text before the match in the original string. Normalized to start with one leading slash and end with zero. """ return "/" + mat.string[: mat.start()].strip("/") def get_repo(backend: "Backend", mat: re.Match[str]) -> BaseRepo: """Get a Repo instance for the given backend and URL regex match.""" return cast(BaseRepo, backend.open_repository(url_prefix(mat))) def send_file( req: "HTTPGitRequest", f: BinaryIO | None, content_type: str ) -> Iterator[bytes]: """Send a file-like object to the request output. Args: req: The HTTPGitRequest object to send output to. f: An open file-like object to send; will be closed. content_type: The MIME type for the file. Returns: Iterator over the contents of the file, as chunks. """ if f is None: yield req.not_found("File not found") return try: req.respond(HTTP_OK, content_type) while True: data = f.read(10240) if not data: break yield data except OSError: yield req.error("Error reading file") finally: f.close() def _url_to_path(url: str) -> str: return url.replace("/", os.path.sep) def get_text_file( req: "HTTPGitRequest", backend: "Backend", mat: re.Match[str] ) -> Iterator[bytes]: """Send a plain text file from the repository. Args: req: The HTTP request object backend: The git backend mat: The regex match for the requested path Returns: Iterator yielding file contents as bytes """ req.nocache() path = _url_to_path(mat.group()) logger.info("Sending plain text file %s", path) return send_file(req, get_repo(backend, mat).get_named_file(path), "text/plain") def get_loose_object( req: "HTTPGitRequest", backend: "Backend", mat: re.Match[str] ) -> Iterator[bytes]: """Send a loose git object. Args: req: The HTTP request object backend: The git backend mat: The regex match containing object path segments Returns: Iterator yielding object contents as bytes """ sha = cast(ObjectID, (mat.group(1) + mat.group(2)).encode("ascii")) logger.info("Sending loose object %s", sha) object_store = get_repo(backend, mat).object_store if not object_store.contains_loose(sha): yield req.not_found("Object not found") return try: data = object_store[sha].as_legacy_object() except OSError: yield req.error("Error reading object") return req.cache_forever() req.respond(HTTP_OK, "application/x-git-loose-object") yield data def get_pack_file( req: "HTTPGitRequest", backend: "Backend", mat: re.Match[str] ) -> Iterator[bytes]: """Send a git pack file. Args: req: The HTTP request object backend: The git backend mat: The regex match for the requested pack file Returns: Iterator yielding pack file contents as bytes """ req.cache_forever() path = _url_to_path(mat.group()) logger.info("Sending pack file %s", path) return send_file( req, get_repo(backend, mat).get_named_file(path), "application/x-git-packed-objects", ) def get_idx_file( req: "HTTPGitRequest", backend: "Backend", mat: re.Match[str] ) -> Iterator[bytes]: """Send a git pack index file. Args: req: The HTTP request object backend: The git backend mat: The regex match for the requested index file Returns: Iterator yielding index file contents as bytes """ req.cache_forever() path = _url_to_path(mat.group()) logger.info("Sending pack file %s", path) return send_file( req, get_repo(backend, mat).get_named_file(path), "application/x-git-packed-objects-toc", ) def get_info_refs( req: "HTTPGitRequest", backend: "Backend", mat: re.Match[str] ) -> Iterator[bytes]: """Send git info/refs for discovery. Args: req: The HTTP request object backend: The git backend mat: The regex match for the info/refs request Returns: Iterator yielding refs advertisement or info/refs contents """ params = parse_qs(req.environ["QUERY_STRING"]) service = params.get("service", [None])[0] try: repo = get_repo(backend, mat) except NotGitRepository as e: yield req.not_found(str(e)) return if service and not req.dumb: if req.handlers is None: yield req.forbidden("No handlers configured") return handler_cls = req.handlers.get(service.encode("ascii"), None) if handler_cls is None: yield req.forbidden("Unsupported service") return req.nocache() write = req.respond(HTTP_OK, f"application/x-{service}-advertisement") def write_fn(data: bytes) -> int | None: result = write(data) return len(data) if result is not None else None proto = ReceivableProtocol(BytesIO().read, write_fn) from typing import Any, cast handler = handler_cls( backend, cast(Any, [url_prefix(mat)]), # handler_cls could expect bytes or str proto, stateless_rpc=True, advertise_refs=True, ) assert handler is not None handler.proto.write_pkt_line(b"# service=" + service.encode("ascii") + b"\n") handler.proto.write_pkt_line(None) handler.handle() else: # non-smart fallback # TODO: select_getanyfile() (see http-backend.c) req.nocache() req.respond(HTTP_OK, "text/plain") logger.info("Emulating dumb info/refs") yield from generate_info_refs(repo) def get_info_packs( req: "HTTPGitRequest", backend: "Backend", mat: re.Match[str] ) -> Iterator[bytes]: """Send git info/packs file listing available packs. Args: req: The HTTP request object backend: The git backend mat: The regex match for the info/packs request Returns: Iterator yielding pack listing as bytes """ req.nocache() req.respond(HTTP_OK, "text/plain") logger.info("Emulating dumb info/packs") return generate_objects_info_packs(get_repo(backend, mat)) # Upper bound on a single dechunked chunk. Matches the largest pkt-line the # smart-http protocol will emit (~64 KiB) with room to spare, and keeps a # hostile ``Transfer-Encoding: chunked`` peer from asking the server to # allocate an arbitrarily large buffer in one call. MAX_CHUNK_SIZE = 1 * 1024 * 1024 # Upper bound on the total dechunked body of a single request. Large enough # to cover realistic pushes but finite so a stream of small chunks can no # longer grow the buffer without limit. Also used by ``LimitedInputFilter`` # as a fallback cap when the request omits ``Content-Length`` (as chunked # requests do). MAX_REQUEST_SIZE = 1 * 1024 * 1024 * 1024 class ChunkedEncodingError(ValueError): """Raised when a chunked-encoded request body is malformed or too large.""" def _chunk_iter( f: BinaryIO, max_chunk_size: int = MAX_CHUNK_SIZE, max_total_size: int = MAX_REQUEST_SIZE, ) -> Iterator[bytes]: total = 0 while True: line = f.readline() if not line: raise ChunkedEncodingError("Truncated chunked stream") # Strip HTTP/1.1 chunk extensions (RFC 7230 section 4.1.1) before # parsing the length. ``int(..., 16)`` accepts a leading ``-``, so # reject negatives explicitly. size_str = line.rstrip(b"\r\n").split(b";", 1)[0].strip() try: length = int(size_str, 16) except ValueError: raise ChunkedEncodingError("Invalid chunk size") from None if length < 0: raise ChunkedEncodingError("Invalid chunk size") if length > max_chunk_size: raise ChunkedEncodingError("Chunk size exceeds maximum allowed") if length == 0: break total += length if total > max_total_size: raise ChunkedEncodingError("Request body exceeds maximum allowed size") chunk = f.read(length + 2) yield chunk[:-2] class ChunkReader: """Reader for chunked transfer encoding streams.""" def __init__(self, f: BinaryIO) -> None: """Initialize ChunkReader. Args: f: Binary file-like object to read from """ self._iter = _chunk_iter(f) self._buffer: list[bytes] = [] self._buffered = 0 def read(self, n: int) -> bytes: """Read n bytes from the chunked stream. Args: n: Number of bytes to read Returns: Up to n bytes of data """ while self._buffered < n: try: chunk = next(self._iter) except StopIteration: break self._buffer.append(chunk) self._buffered += len(chunk) f = b"".join(self._buffer) ret = f[:n] self._buffer = [f[n:]] self._buffered = len(f) - len(ret) return ret class _LengthLimitedFile: """Wrapper class to limit the length of reads from a file-like object. This is used to ensure EOF is read from the wsgi.input object once Content-Length bytes are read. This behavior is required by the WSGI spec but not implemented in wsgiref as of 2.5. """ def __init__(self, input: BinaryIO, max_bytes: int) -> None: self._input = input self._bytes_avail = max_bytes def read(self, size: int = -1) -> bytes: """Read up to size bytes from the limited input. Args: size: Maximum number of bytes to read, or -1 for all available Returns: Up to size bytes of data """ if self._bytes_avail <= 0: return b"" if size == -1 or size > self._bytes_avail: size = self._bytes_avail self._bytes_avail -= size return self._input.read(size) def readline(self, size: int = -1) -> bytes: """Read a single line, honouring the remaining byte budget.""" if self._bytes_avail <= 0: return b"" if size == -1 or size > self._bytes_avail: size = self._bytes_avail line = self._input.readline(size) self._bytes_avail -= len(line) return line # TODO: support more methods as necessary def handle_service_request( req: "HTTPGitRequest", backend: "Backend", mat: re.Match[str] ) -> Iterator[bytes]: """Handle a git service request (upload-pack or receive-pack). Args: req: The HTTP request object backend: The git backend mat: The regex match for the service request Returns: Iterator yielding service response as bytes """ service = mat.group().lstrip("/") logger.info("Handling service request for %s", service) # Require the exact Git smart HTTP request Content-Type, matching # ``git http-backend``. The value is not a CORS-safe MIME type, so browsers # must preflight cross-origin POSTs. expected_content_type = f"application/x-{service}-request" if req.environ.get("CONTENT_TYPE") != expected_content_type: yield req.forbidden("Invalid Content-Type") return if req.handlers is None: yield req.forbidden("No handlers configured") return handler_cls = req.handlers.get(service.encode("ascii"), None) if handler_cls is None: yield req.forbidden("Unsupported service") return try: get_repo(backend, mat) except NotGitRepository as e: yield req.not_found(str(e)) return req.nocache() write = req.respond(HTTP_OK, f"application/x-{service}-result") def write_fn(data: bytes) -> int | None: result = write(data) return len(data) if result is not None else None if req.environ.get("HTTP_TRANSFER_ENCODING") == "chunked": read = ChunkReader(req.environ["wsgi.input"]).read else: read = req.environ["wsgi.input"].read proto = ReceivableProtocol(read, write_fn) # TODO(jelmer): Find a way to pass in repo, rather than having handler_cls # reopen. handler = handler_cls( backend, [url_prefix(mat).encode("utf-8")], proto, stateless_rpc=True ) assert handler is not None handler.handle() class HTTPGitRequest: """Class encapsulating the state of a single git HTTP request. Attributes: environ: the WSGI environment for the request. """ def __init__( self, environ: WSGIEnvironment, start_response: StartResponse, dumb: bool = False, handlers: dict[bytes, "HandlerConstructor | Callable[..., Any]"] | None = None, ) -> None: """Initialize HTTPGitRequest. Args: environ: WSGI environment dictionary start_response: WSGI start_response callable dumb: Whether to use dumb HTTP protocol handlers: Optional handler overrides """ self.environ = environ self.dumb = dumb self.handlers = handlers self._start_response = start_response self._cache_headers: list[tuple[str, str]] = [] self._headers: list[tuple[str, str]] = [] def add_header(self, name: str, value: str) -> None: """Add a header to the response.""" self._headers.append((name, value)) def respond( self, status: str = HTTP_OK, content_type: str | None = None, headers: Sequence[tuple[str, str]] | None = None, ) -> Callable[[bytes], object]: """Begin a response with the given status and other headers.""" if headers: self._headers.extend(headers) if content_type: self._headers.append(("Content-Type", content_type)) self._headers.extend(self._cache_headers) return self._start_response(status, self._headers) def not_found(self, message: str) -> bytes: """Begin a HTTP 404 response and return the text of a message.""" self._cache_headers = [] logger.info("Not found: %s", message) self.respond(HTTP_NOT_FOUND, "text/plain") return message.encode("ascii") def forbidden(self, message: str) -> bytes: """Begin a HTTP 403 response and return the text of a message.""" self._cache_headers = [] logger.info("Forbidden: %s", message) self.respond(HTTP_FORBIDDEN, "text/plain") return message.encode("ascii") def error(self, message: str) -> bytes: """Begin a HTTP 500 response and return the text of a message.""" self._cache_headers = [] logger.error("Error: %s", message) self.respond(HTTP_ERROR, "text/plain") return message.encode("ascii") def nocache(self) -> None: """Set the response to never be cached by the client.""" self._cache_headers = NO_CACHE_HEADERS def cache_forever(self) -> None: """Set the response to be cached forever by the client.""" self._cache_headers = cache_forever_headers() class HTTPGitApplication: """Class encapsulating the state of a git WSGI application. Attributes: backend: the Backend object backing this application """ services: ClassVar[ dict[ tuple[str, re.Pattern[str]], Callable[[HTTPGitRequest, Backend, re.Match[str]], Iterator[bytes]], ] ] = { ("GET", re.compile("/HEAD\\Z")): get_text_file, ("GET", re.compile("/info/refs\\Z")): get_info_refs, ("GET", re.compile("/objects/info/alternates\\Z")): get_text_file, ("GET", re.compile("/objects/info/http-alternates\\Z")): get_text_file, ("GET", re.compile("/objects/info/packs\\Z")): get_info_packs, ( "GET", re.compile("/objects/([0-9a-f]{2})/([0-9a-f]{38})\\Z"), ): get_loose_object, # Match any "-" pack basename, not just "pack-". ``git # maintenance`` writes "loose-" packs, which get_info_packs # advertises. The prefix is restricted to word characters and the # hash to hex so the matched name can never contain a path separator # (get_named_file joins it under the control dir unsanitised). ( "GET", re.compile("/objects/pack/\\w+-([0-9a-f]{40}|[0-9a-f]{64})\\.pack\\Z"), ): get_pack_file, ( "GET", re.compile("/objects/pack/\\w+-([0-9a-f]{40}|[0-9a-f]{64})\\.idx\\Z"), ): get_idx_file, ("POST", re.compile("/git-upload-pack\\Z")): handle_service_request, ("POST", re.compile("/git-receive-pack\\Z")): handle_service_request, } def __init__( self, backend: Backend, dumb: bool = False, handlers: dict[bytes, "HandlerConstructor | Callable[..., Any]"] | None = None, fallback_app: WSGIApplication | None = None, ) -> None: """Initialize HTTPGitApplication. Args: backend: Backend object for git operations dumb: Whether to use dumb HTTP protocol handlers: Optional handler overrides fallback_app: Optional fallback WSGI application """ self.backend = backend self.dumb = dumb self.handlers: dict[bytes, HandlerConstructor | Callable[..., Any]] = dict( DEFAULT_HANDLERS ) self.fallback_app = fallback_app if handlers is not None: self.handlers.update(handlers) def __call__( self, environ: WSGIEnvironment, start_response: StartResponse, ) -> Iterable[bytes]: """Handle WSGI request.""" path = environ["PATH_INFO"] method = environ["REQUEST_METHOD"] req = HTTPGitRequest( environ, start_response, dumb=self.dumb, handlers=self.handlers ) # environ['QUERY_STRING'] has qs args handler = None mat = None for smethod, spath in self.services.keys(): if smethod != method: continue mat = spath.search(path) if mat: # url_prefix() collapses slashes, so "/repo//git-receive-pack" # would otherwise resolve to the same repository as the # canonical spelling. Reject the redundant form rather than # normalise it, so only one path reaches a given repository. if mat.string[: mat.start()].endswith("/"): mat = None continue handler = self.services[smethod, spath] break if handler is None or mat is None: if self.fallback_app is not None: return self.fallback_app(environ, start_response) else: return [req.not_found("Sorry, that method is not supported")] return handler(req, self.backend, mat) class GunzipFilter: """WSGI middleware that unzips gzip-encoded requests before passing on to the underlying application.""" def __init__(self, application: WSGIApplication) -> None: """Initialize GunzipFilter with WSGI application.""" self.app = application def __call__( self, environ: WSGIEnvironment, start_response: StartResponse, ) -> Iterable[bytes]: """Handle WSGI request with gzip decompression.""" import gzip if environ.get("HTTP_CONTENT_ENCODING", "") == "gzip": environ["wsgi.input"] = gzip.GzipFile( filename=None, fileobj=environ["wsgi.input"], mode="rb" ) del environ["HTTP_CONTENT_ENCODING"] environ.pop("CONTENT_LENGTH", None) return self.app(environ, start_response) class LimitedInputFilter: """WSGI middleware that limits the input length of a request. When ``Content-Length`` is present the input is capped at that value. Requests that omit ``Content-Length`` (notably HTTP/1.1 chunked transfer-encoded requests) are still capped at ``MAX_REQUEST_SIZE`` bytes so a hostile peer cannot force the server to buffer an unbounded body. """ def __init__( self, application: WSGIApplication, max_request_size: int = MAX_REQUEST_SIZE, ) -> None: """Initialize LimitedInputFilter with WSGI application.""" self.app = application self.max_request_size = max_request_size def __call__( self, environ: WSGIEnvironment, start_response: StartResponse, ) -> Iterable[bytes]: """Handle WSGI request with input length limiting.""" # This is not necessary if this app is run from a conforming WSGI # server. Unfortunately, there's no way to tell that at this point. content_length = environ.get("CONTENT_LENGTH", "") if content_length: try: limit = int(content_length) except ValueError: limit = self.max_request_size else: if limit < 0 or limit > self.max_request_size: limit = self.max_request_size else: limit = self.max_request_size environ["wsgi.input"] = _LengthLimitedFile(environ["wsgi.input"], limit) return self.app(environ, start_response) def make_wsgi_chain( backend: Backend, dumb: bool = False, handlers: dict[bytes, Callable[..., Any]] | None = None, fallback_app: WSGIApplication | None = None, ) -> WSGIApplication: """Factory function to create an instance of HTTPGitApplication. Correctly wrapped with needed middleware. """ app = HTTPGitApplication( backend, dumb=dumb, handlers=handlers, fallback_app=fallback_app ) wrapped_app = LimitedInputFilter(GunzipFilter(app)) return wrapped_app class ServerHandlerLogger(ServerHandler): """ServerHandler that uses dulwich's logger for logging exceptions.""" def log_exception( self, exc_info: tuple[type[BaseException], BaseException, TracebackType] | tuple[None, None, None] | None, ) -> None: """Log exception using dulwich logger.""" logger.exception( "Exception happened during processing of request", exc_info=exc_info, ) def log_message(self, format: str, *args: object) -> None: """Log message using dulwich logger.""" logger.info(format, *args) def log_error(self, format: str, *args: object) -> None: """Log error using dulwich logger.""" logger.error(format, *args) class WSGIRequestHandlerLogger(WSGIRequestHandler): """WSGIRequestHandler that uses dulwich's logger for logging exceptions.""" def log_exception( self, exc_info: tuple[type[BaseException], BaseException, TracebackType] | tuple[None, None, None] | None, ) -> None: """Log exception using dulwich logger.""" logger.exception( "Exception happened during processing of request", exc_info=exc_info, ) def log_message(self, format: str, *args: object) -> None: """Log message using dulwich logger.""" logger.info(format, *args) def log_error(self, format: str, *args: object) -> None: """Log error using dulwich logger.""" logger.error(format, *args) def handle(self) -> None: """Handle a single HTTP request.""" self.raw_requestline = self.rfile.readline() if not self.parse_request(): # An error code has been sent, just exit return handler = ServerHandlerLogger( self.rfile, self.wfile, # type: ignore self.get_stderr(), self.get_environ(), ) handler.request_handler = self # type: ignore # backpointer for logging handler.run(self.server.get_app()) # type: ignore class WSGIServerLogger(WSGIServer): """WSGIServer that uses dulwich's logger for error handling.""" def handle_error(self, request: object, client_address: tuple[str, int]) -> None: """Handle an error.""" logger.exception( f"Exception happened during processing of request from {client_address!s}" ) def main(argv: list[str] = sys.argv) -> None: """Entry point for starting an HTTP git server.""" import optparse parser = optparse.OptionParser() parser.add_option( "-l", "--listen_address", dest="listen_address", default="localhost", help="Binding IP address.", ) parser.add_option( "-p", "--port", dest="port", type=int, default=8000, help="Port to listen on.", ) options, args = parser.parse_args(argv) if len(args) > 1: gitdir = args[1] else: gitdir = os.getcwd() log_utils.default_logging_config() from typing import cast from dulwich.server import BackendRepo backend = DictBackend({"/": cast(BackendRepo, Repo(gitdir))}) app = make_wsgi_chain(backend) server = make_server( options.listen_address, options.port, app, handler_class=WSGIRequestHandlerLogger, server_class=WSGIServerLogger, ) logger.warning( "This server is intended for debugging only. " "For production use, run the WSGI app returned by make_wsgi_chain() " "inside a proper WSGI server (e.g. gunicorn, uWSGI, mod_wsgi). " "The GunzipFilter middleware it wraps is required, as WSGI servers " "do not decode gzip-encoded request bodies that git clients send." ) logger.info( "Listening for HTTP connections on %s:%d", options.listen_address, options.port, ) server.serve_forever() if __name__ == "__main__": main()