diff --git a/app.py b/app.py index 2fc29fa..2b14377 100644 --- a/app.py +++ b/app.py @@ -25,6 +25,12 @@ from flask import ( Response, ) from flask_compress import Compress +from source_conversion import ( + InvalidSourceError, + SourceTooLargeError, + convert_source, + iter_completed_bytes, +) from flask_limiter import Limiter import config @@ -381,31 +387,18 @@ def index(): @app.route("/convert") @limiter.limit(config.RATE_LIMIT_CONVERT) def convert(): - - - def is_private_client_ip() -> bool: - ip = get_client_ip() - try: - return ipaddress.ip_address(ip).is_private - except Exception: - return False + # Flask automatically maps HEAD to a GET route as well. + if request.method == "HEAD": + return convert_head() requested_debug = request.args.get("debug", "").lower() in ( - "1", - "true", - "t", - "yes", - "y", - "on", + "1", "true", "t", "yes", "y", "on", ) debug_allowed = False if config.DEBUG_ENABLE: header_key = request.headers.get("X-Debug-Key", "") - if ( - config.DEBUG_KEY - and header_key - and hmac.compare_digest(header_key, config.DEBUG_KEY) - ): + if (config.DEBUG_KEY and header_key + and hmac.compare_digest(header_key, config.DEBUG_KEY)): debug_allowed = True elif is_private_client_ip(): debug_allowed = True @@ -417,182 +410,160 @@ def convert(): debug_lines = [] def d(msg): - ts = datetime.now().isoformat() - line = f"# [DEBUG {ts}] {msg}" + line = f"# [DEBUG {datetime.now().isoformat()}] {msg}" debug_lines.append(line) app.logger.debug(line) def debug_response(status=200): body = "\n".join(debug_lines) + ("\n" if debug_lines else "") - resp = Response(body, mimetype="text/plain; charset=utf-8", status=status) + resp = Response(body, content_type="text/plain; charset=utf-8", status=status) resp.headers["X-Debug-Mode"] = "1" resp.headers["Cache-Control"] = "no-store" return resp + def reject(status, description): + redis_client.incr(f"stats:errors_{status}") + if debug_mode: + d(description) + return debug_response(status=status) + abort(status, description=description) + + add_recent_convert() + redis_client.incr("stats:convert_requests") + if debug_mode: + d("Start /convert w trybie debug") + + # request.args already URL-decodes the outer query parameter. Calling + # unquote() again corrupts percent escapes within the nested source URL. + source_url = request.args.get("url") + if not source_url: + return reject(400, "Brak parametru url z adresem listy.") try: - add_recent_convert() - redis_client.incr("stats:convert_requests") - if debug_mode: - d("Start /convert w trybie debug") + normalized_url = validate_and_normalize_url(source_url) + except ValueError: + return reject(400, "Nieprawid\u0142owy adres URL listy.") - encoded_url = request.args.get("url") - if not encoded_url: - if debug_mode: - d("Brak parametru ?url") - return debug_response(status=400) - redis_client.incr("stats:errors_400") - abort(400, description="Missing URL parameter") + try: + target_ip = validate_ip(request.args.get("ip", "127.0.0.1")) + except ValueError: + return reject(400, "Nieprawid\u0142owy adres IP.") - decoded_url = unquote(encoded_url) - try: - normalized_url = validate_and_normalize_url(decoded_url) - except ValueError as e: - if debug_mode: - d(f"Błąd walidacji URL: {e}") - return debug_response(status=400) - redis_client.incr("stats:errors_400") - abort(400) + if debug_mode: + d(f"URL: {normalized_url}") + d(f"target_ip: {target_ip}") - try: - target_ip = validate_ip(request.args.get("ip", "127.0.0.1")) - except ValueError: - if debug_mode: - d("Bad parametr ?ip") - return debug_response(status=400) - redis_client.incr("stats:errors_400") - abort(400, description="Invalid IP") + track_url_request(normalized_url) + redis_client.incr(f"stats:target_ips:{target_ip}") - if debug_mode: - d(f"URL (encoded): {encoded_url}") - d(f"URL (decoded): {decoded_url}") - d(f"URL (norm): {normalized_url}") - d(f"target_ip: {target_ip}") + req_headers = {} + for client_header in ("If-None-Match", "If-Modified-Since"): + value = request.headers.get(client_header) + if value: + req_headers[client_header] = value - track_url_request(normalized_url) - redis_client.incr(f"stats:target_ips:{target_ip}") - - # nagłówki If-* - req_headers = {} - inm = request.headers.get("If-None-Match") - ims = request.headers.get("If-Modified-Since") - if inm: - req_headers["If-None-Match"] = inm - if ims: - req_headers["If-Modified-Since"] = ims - if debug_mode: - d("Wysyłam GET do upstreamu") - d(f"Nagłówki: {req_headers or '{}'}") - - r = requests.get( + try: + upstream = requests.get( normalized_url, headers=req_headers, stream=True, timeout=(10, 60) ) + except requests.exceptions.RequestException as exc: + app.logger.warning("Cannot fetch source list: %s", exc) + return reject(502, "Nie mo\u017cna pobra\u0107 listy z podanego adresu.") - ct = r.headers.get("Content-Type", "") + result = None + try: if debug_mode: - d(f"Upstream status: {r.status_code}") - d(f"Content-Type: {ct or '(brak)'}") - d(f"ETag: {r.headers.get('ETag')}") - d(f"Last-Modified: {r.headers.get('Last-Modified')}") + d(f"Upstream HTTP: {upstream.status_code}") + d(f"Content-Type: {upstream.headers.get('Content-Type', '(brak)')}") - if "text" not in ct and "octet-stream" not in ct and ct != "": - if debug_mode: - d("Unsupported Media Type -> 415") - r.close() - return debug_response(status=415) - r.close() - abort(415, description="Unsupported Media Type") - - if r.status_code == 304: + if upstream.status_code == 304: etag = build_etag( - r.headers.get("ETag"), r.headers.get("Last-Modified"), target_ip + upstream.headers.get("ETag"), + upstream.headers.get("Last-Modified"), + target_ip, ) if debug_mode: - d("Upstream 304 – zwracam 304") - r.close() - add_recent_link(normalized_url, target_ip) + d("Upstream HTTP 304") return debug_response(status=304) resp = Response(status=304) - resp.headers.update(cache_headers(etag, r.headers.get("Last-Modified"))) - resp.direct_passthrough = True - r.close() + resp.headers.update(cache_headers(etag, upstream.headers.get("Last-Modified"))) add_recent_link(normalized_url, target_ip) return resp - up_etag = r.headers.get("ETag") - up_lm = r.headers.get("Last-Modified") - etag = build_etag(up_etag, up_lm, target_ip) - if debug_mode: - d(f"Etag dla klienta: {etag}") + if upstream.status_code != 200: + return reject( + 502, + f"Nie mo\u017cna pobra\u0107 listy. Serwer \u017ar\u00f3d\u0142owy zwr\u00f3ci\u0142 HTTP {upstream.status_code}.", + ) - @stream_with_context - def body_gen(): - lines_read = 0 - lines_emitted = 0 - try: - if debug_mode: - yield "\n".join(debug_lines) + "\n" - debug_lines.clear() - for line in r.iter_lines( - decode_unicode=True, chunk_size=config.READ_CHUNK - ): - if line is None: - continue - lines_read += 1 - if len(line) > config.STREAM_LINE_LIMIT: - if debug_mode and lines_read <= 5: - yield f"# [DEBUG] pominięto długi wiersz ({len(line)} bajtów)\n" - continue - out = convert_host_line(line, target_ip) - if out: - lines_emitted += 1 - yield out + "\n" - if debug_mode and lines_read <= 5: - preview = line[:200].replace("\r", "\\r").replace("\n", "\\n") - yield f"# [DEBUG] podgląd linii {lines_read}: {preview}\n" - if debug_mode: - yield f"# [DEBUG] podsumowanie: przeczytano={lines_read}, wyemitowano={lines_emitted}\n" - if lines_emitted == 0: - yield "# [DEBUG] Uwaga: 0 linii wynikowych – czy format listy pasuje?\n" - redis_client.incrby("stats:content_size_total", 0) - redis_client.incr("stats:content_size_count") - finally: - r.close() + content_type = upstream.headers.get("Content-Type", "") + media_type = content_type.split(";", 1)[0].strip().lower() + if (media_type and ( + media_type in {"text/html", "application/xhtml+xml"} + or not (media_type.startswith("text/") or media_type == "application/octet-stream") + )): + return reject(415, "Podany adres nie zwraca obs\u0142ugiwanej listy tekstowej.") - resp = Response(body_gen(), mimetype="text/plain; charset=utf-8") - resp.headers.update(cache_headers(etag, up_lm)) - resp.direct_passthrough = True - redis_client.incr("stats:conversions_success") - add_recent_link(normalized_url, target_ip) - return resp + try: + result = convert_source( + upstream, + target_ip, + convert_host_line, + read_chunk=config.READ_CHUNK, + line_limit=config.STREAM_LINE_LIMIT, + source_limit=config.MAX_SOURCE_BYTES, + output_limit=config.MAX_CONVERTED_BYTES, + debug_lines=debug_lines if debug_mode else None, + ) + except SourceTooLargeError as exc: + app.logger.info("Source or output size limit exceeded: %s", exc) + return reject(413, "Plik listy jest zbyt du\u017cy do przetworzenia.") + except InvalidSourceError as exc: + app.logger.info("Unsupported or corrupted source list: %s", exc) + return reject( + 422, + "Nie mo\u017cna zdekodowa\u0107 ani obs\u0142u\u017cy\u0107 tej listy. " + "Sprawd\u017a adres URL oraz format pliku.", + ) + except requests.exceptions.RequestException as exc: + app.logger.warning("Error while reading source body: %s", exc) + return reject(502, "Przerwano pobieranie listy z serwera \u017ar\u00f3d\u0142owego.") + finally: + upstream.close() - except requests.exceptions.RequestException as e: - app.logger.error(f"Request error: {str(e)}") - redis_client.incr("stats:errors_500") - if debug_mode: - d(f"Wyjątek requests: {e}") - return debug_response(status=502) - abort(500) - except ValueError as e: - app.logger.error(f"URL validation error: {str(e)}") - redis_client.incr("stats:errors_400") - if debug_mode: - d(f"Wyjątek ValueError: {e}") - return debug_response(status=400) - abort(400) + up_etag = upstream.headers.get("ETag") + up_lm = upstream.headers.get("Last-Modified") + etag = build_etag(up_etag, up_lm, target_ip) + + # The entire response was decoded, validated and spooled successfully. + # In particular, this iterator produces BYTES, as required by WSGI/ASGI. + resp = Response( + iter_completed_bytes(result.output, config.READ_CHUNK), + content_type="text/plain; charset=utf-8", + ) + resp.headers.update(cache_headers(etag, up_lm)) + resp.headers["Content-Length"] = str(result.output_size) + resp.direct_passthrough = True + resp.call_on_close(result.output.close) + redis_client.incr("stats:conversions_success") + redis_client.incrby("stats:content_size_total", result.output_size) + redis_client.incr("stats:content_size_count") + add_recent_link(normalized_url, target_ip) + return resp @app.route("/convert", methods=["HEAD"]) def convert_head(): - encoded_url = request.args.get("url", config.DEFAULT_SOURCE_URL) - if not encoded_url: - abort(400) - decoded_url = unquote(encoded_url) - validate_and_normalize_url(decoded_url) - target_ip = validate_ip(request.args.get("ip", "127.0.0.1")) + source_url = request.args.get("url", config.DEFAULT_SOURCE_URL) + if not source_url: + abort(400, description="Brak parametru url z adresem listy.") + try: + validate_and_normalize_url(source_url) + target_ip = validate_ip(request.args.get("ip", "127.0.0.1")) + except ValueError: + abort(400, description="Nieprawid\u0142owy adres URL lub IP.") etag = build_etag(None, None, target_ip) resp = Response(status=200) resp.headers.update(cache_headers(etag, None)) - resp.direct_passthrough = True return resp @@ -770,6 +741,8 @@ def stats_json(): @app.errorhandler(404) @app.errorhandler(413) @app.errorhandler(415) +@app.errorhandler(422) +@app.errorhandler(502) @app.errorhandler(500) def handle_errors(e): try: diff --git a/config.py b/config.py index 21752b8..0b587e3 100644 --- a/config.py +++ b/config.py @@ -53,6 +53,8 @@ AIOHTTP_SOCK_READ_TIMEOUT = getenv_float("AIOHTTP_SOCK_READ_TIMEOUT", 60.0) READ_CHUNK = getenv_int("READ_CHUNK", 64 * 1024) # 64 KiB STREAM_LINE_LIMIT= getenv_int("STREAM_LINE_LIMIT", 4096) +MAX_SOURCE_BYTES= getenv_int("MAX_SOURCE_BYTES", 100 * 1024 * 1024) +MAX_CONVERTED_BYTES= getenv_int("MAX_CONVERTED_BYTES", 100 * 1024 * 1024) # Serwer BIND_HOST = os.getenv("BIND_HOST", "127.0.0.1") diff --git a/source_conversion.py b/source_conversion.py new file mode 100644 index 0000000..af6b370 --- /dev/null +++ b/source_conversion.py @@ -0,0 +1,181 @@ +"""Validate and convert complete upstream lists before returning HTTP 200. + +The output is spooled to disk when it grows, avoiding a large in-memory +response. Validation and download errors happen before response headers are +sent to the browser or reverse proxy. +""" + +import codecs +import re +import tempfile +from dataclasses import dataclass +from typing import BinaryIO, Callable + + +class InvalidSourceError(Exception): + """An upstream body is not a supported, decodable hosts list.""" + + +class SourceTooLargeError(InvalidSourceError): + """The source or converted result exceeds a configured limit.""" + + +@dataclass +class ConversionResult: + output: BinaryIO + lines_read: int + lines_emitted: int + output_size: int + + +def _get_encoding(content_type: str) -> str: + match = re.search( + r'(?:^|;)\s*charset\s*=\s*(?:"([^"]+)"|([^;\s]+))', + content_type or "", re.IGNORECASE, + ) + encoding = (match.group(1) or match.group(2)) if match else "utf-8-sig" + try: + encoding = codecs.lookup(encoding).name + except (LookupError, ValueError) as exc: + raise InvalidSourceError("Nieznane kodowanie tekstu z serwera zrodlowego.") from exc + # The reader splits on ASCII newline octets. Multi-byte encodings need a + # different framing protocol and must not be interpreted as hosts text. + if encoding in {"utf-16", "utf-16-le", "utf-16-be", "utf-32", "utf-32-le", "utf-32-be"}: + raise InvalidSourceError("Nieobslugiwane kodowanie pliku listy.") + return encoding + + +def _bounded_lines(response, *, chunk_size: int, line_limit: int, source_limit: int): + """Return raw lines, or None for overlong lines; never buffer a huge line.""" + pending = bytearray() + skipping = False + total = 0 + + for chunk in response.iter_content(chunk_size=chunk_size): + if not chunk: + continue + if not isinstance(chunk, bytes): + raise InvalidSourceError("Zrodlo nie zwrocilo poprawnych danych binarnych.") + total += len(chunk) + if total > source_limit: + raise SourceTooLargeError("Plik zrodlowy przekracza dopuszczalny rozmiar.") + pieces = chunk.split(b"\n") + for part in pieces[:-1]: + if skipping or len(pending) + len(part) > line_limit: + yield None + else: + pending.extend(part) + yield bytes(pending).rstrip(b"\r") + pending.clear() + skipping = False + last = pieces[-1] + if not skipping: + if len(pending) + len(last) > line_limit: + pending.clear() + skipping = True + else: + pending.extend(last) + + if skipping: + yield None + elif pending: + yield bytes(pending).rstrip(b"\r") + + +def convert_source( + response, + target_ip: str, + convert_line: Callable[[str, str], str | None], + *, + read_chunk: int, + line_limit: int, + source_limit: int, + output_limit: int, + spool_memory_limit: int = 2 * 1024 * 1024, + debug_lines: list[str] | None = None, +) -> ConversionResult: + """Consume, decode and check the whole source before returning output. + + The caller owns and must close result.output. On failure, the spool is + closed here, so no partial result can be published or cached as valid. + """ + encoding = _get_encoding(response.headers.get("Content-Type", "")) + length = response.headers.get("Content-Length", "") + if length: + try: + if int(length) > source_limit: + raise SourceTooLargeError("Plik zrodlowy przekracza dopuszczalny rozmiar.") + except ValueError: + pass + + output = tempfile.SpooledTemporaryFile(max_size=spool_memory_limit, mode="w+b") + lines_read = 0 + lines_emitted = 0 + written = 0 + + def append(data: bytes): + nonlocal written + if written + len(data) > output_limit: + raise SourceTooLargeError("Wynik konwersji przekracza dopuszczalny rozmiar.") + output.write(data) + written += len(data) + + try: + if debug_lines is not None and debug_lines: + append(("\n".join(debug_lines) + "\n").encode("utf-8")) + + for raw in _bounded_lines( + response, + chunk_size=read_chunk, + line_limit=line_limit, + source_limit=source_limit, + ): + lines_read += 1 + if raw is None: + if debug_lines is not None and lines_read <= 5: + append(b"# [DEBUG] pominieto zbyt dlugi wiersz\n") + continue + if b"\x00" in raw: + raise InvalidSourceError("Zrodlo zawiera dane binarne zamiast listy tekstowej.") + try: + line = raw.decode(encoding, errors="strict") + if lines_read == 1: + line = line.lstrip("\ufeff") + except UnicodeError as exc: + raise InvalidSourceError("Nie mozna zdekodowac tekstu listy (kodowanie).") from exc + try: + converted = convert_line(line, target_ip) + except (TypeError, ValueError) as exc: + raise InvalidSourceError("Nie mozna przetworzyc formatu listy.") from exc + if converted: + lines_emitted += 1 + append((converted + "\n").encode("utf-8")) + if debug_lines is not None and lines_read <= 5: + preview = line[:200].replace("\r", "\\r").replace("\n", "\\n") + append(f"# [DEBUG] podglad {lines_read}: {preview}\n".encode("utf-8")) + + if lines_emitted == 0: + raise InvalidSourceError( + "Nie znaleziono domen w obslugiwanym formacie. Sprawdz adres i zawartosc listy." + ) + if debug_lines is not None: + append( + f"# [DEBUG] podsumowanie: przeczytano={lines_read}, wyemitowano={lines_emitted}\n".encode("utf-8") + ) + output.seek(0) + return ConversionResult(output, lines_read, lines_emitted, written) + except BaseException: + output.close() + raise + + +def iter_completed_bytes(output: BinaryIO, chunk_size: int): + """The WSGI iterable must produce bytes, especially for WsgiToAsgi.""" + try: + while True: + data = output.read(chunk_size) + if not data: + break + yield data + finally: + output.close() diff --git a/templates/error.html b/templates/error.html index 0c9ffc6..bec1bb0 100644 --- a/templates/error.html +++ b/templates/error.html @@ -36,7 +36,9 @@ {% elif (error.code or 500) == 403 %}Forbidden {% elif (error.code or 500) == 404 %}Not found {% elif (error.code or 500) == 413 %}Payload too large - {% elif (error.code or 500) == 415 %}Unsupported media type + {% elif (error.code or 500) == 415 %}Nieobsługiwany format listy + {% elif (error.code or 500) == 422 %}Nie można przetworzyć listy + {% elif (error.code or 500) == 502 %}Nie udało się pobrać listy {% elif (error.code or 500) == 500 %}Internal server error {% else %}Something went wrong {% endif %} diff --git a/tests/test_source_conversion.py b/tests/test_source_conversion.py new file mode 100644 index 0000000..ae8e640 --- /dev/null +++ b/tests/test_source_conversion.py @@ -0,0 +1,122 @@ +import unittest + +from source_conversion import ( + InvalidSourceError, + SourceTooLargeError, + convert_source, + iter_completed_bytes, +) + + +class FakeUpstream: + def __init__(self, body, headers=None, fail_after=None): + self.body = body + self.headers = headers or {"Content-Type": "text/plain; charset=utf-8"} + self.fail_after = fail_after + + def iter_content(self, chunk_size): + for i in range(0, len(self.body), chunk_size): + yield self.body[i:i + chunk_size] + if self.fail_after is not None and i >= self.fail_after: + raise IOError("upstream disconnected") + + +def convert_test_line(line, ip): + line = line.strip() + if not line or line.startswith('#'): + return None + if line.startswith('||') and line.endswith('^'): + return f'{ip} {line[2:-1]}' + if line.startswith('0.0.0.0 '): + return f'{ip} {line.split()[1]}' + if line.endswith('.test'): + return f'{ip} {line}' + return None + + +def run_conversion(body, *, headers=None, chunk=7, source_limit=1024, output_limit=1024, + line_limit=128, debug=None, fail_after=None): + return convert_source( + FakeUpstream(body, headers=headers, fail_after=fail_after), + "0.0.0.0", + convert_test_line, + read_chunk=chunk, + line_limit=line_limit, + source_limit=source_limit, + output_limit=output_limit, + spool_memory_limit=16, + debug_lines=debug, + ) + + +class ConvertSourceTests(unittest.TestCase): + def test_valid_host_list_is_bytes_and_completed(self): + src = b'# comment\n0.0.0.0 first.test\n||second.test^\n' + result = run_conversion(src) + body_chunks = list(iter_completed_bytes(result.output, 5)) + self.assertTrue(all(isinstance(v, bytes) for v in body_chunks)) + self.assertEqual(b''.join(body_chunks), b'0.0.0.0 first.test\n0.0.0.0 second.test\n') + self.assertTrue(result.output.closed) + self.assertEqual(result.lines_emitted, 2) + + def test_octet_stream_without_charset(self): + result = run_conversion( + b'example.test\n', headers={"Content-Type": "application/octet-stream"} + ) + self.assertEqual(result.output.read(), b'0.0.0.0 example.test\n') + result.output.close() + + def test_utf8_bom(self): + result = run_conversion(b'\xef\xbb\xbfexample.test\n') + self.assertEqual(result.output.read(), b'0.0.0.0 example.test\n') + result.output.close() + + def test_no_supported_domain_returns_error(self): + with self.assertRaises(InvalidSourceError): + run_conversion(b'\n