docker/monitoring/mtls-bridge/app.py

import logging
import os
import time

import requests
from flask import Flask, Response, g, request

app = Flask(__name__)

logging.basicConfig(
    level=os.environ.get("LOG_LEVEL", "INFO"),
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger("mtls-bridge")
logging.getLogger("werkzeug").setLevel(logging.WARNING)

# Config via env
TARGET_URL = os.environ.get("TARGET_URL")
CLIENT_CERT = os.environ.get("CLIENT_CERT", "/certs/client.crt")
CLIENT_KEY = os.environ.get("CLIENT_KEY", "/certs/client.key")
UPSTREAM_CA_CERT = os.environ.get("UPSTREAM_CA_CERT", os.environ.get("CA_CERT", "")).strip()
TIMEOUT = int(os.environ.get("TIMEOUT", "5"))

logger.info(
    "mtls-bridge starting target_url=%s timeout=%ss cert=%s key=%s ca=%s log_level=%s",
    TARGET_URL,
    TIMEOUT,
    CLIENT_CERT,
    CLIENT_KEY,
    CA_CERT,
    os.environ.get("LOG_LEVEL", "INFO"),
)


def get_verify_setting():
    if not UPSTREAM_CA_CERT:
        return True

    lowered = UPSTREAM_CA_CERT.lower()
    if lowered in {"false", "0", "no"}:
        logger.warning("TLS verification for upstream is disabled via UPSTREAM_CA_CERT=%s", UPSTREAM_CA_CERT)
        return False

    if not os.path.exists(UPSTREAM_CA_CERT):
        logger.warning(
            "Configured UPSTREAM_CA_CERT path does not exist: %s (falling back to system CA bundle)",
            UPSTREAM_CA_CERT,
        )
        return True

    return UPSTREAM_CA_CERT


VERIFY_SETTING = get_verify_setting()

logger.info(
    "mtls-bridge starting target_url=%s timeout=%ss cert=%s key=%s verify=%s log_level=%s",
    TARGET_URL,
    TIMEOUT,
    CLIENT_CERT,
    CLIENT_KEY,
    VERIFY_SETTING,
    os.environ.get("LOG_LEVEL", "INFO"),
)


def get_verify_setting():
    if not UPSTREAM_CA_CERT:
        return True

    lowered = UPSTREAM_CA_CERT.lower()
    if lowered in {"false", "0", "no"}:
        logger.warning("TLS verification for upstream is disabled via UPSTREAM_CA_CERT=%s", UPSTREAM_CA_CERT)
        return False

    if not os.path.exists(UPSTREAM_CA_CERT):
        logger.warning(
            "Configured UPSTREAM_CA_CERT path does not exist: %s (falling back to system CA bundle)",
            UPSTREAM_CA_CERT,
        )
        return True

    return UPSTREAM_CA_CERT


VERIFY_SETTING = get_verify_setting()

logger.info(
    "mtls-bridge starting target_url=%s timeout=%ss cert=%s key=%s verify=%s log_level=%s",
    TARGET_URL,
    TIMEOUT,
    CLIENT_CERT,
    CLIENT_KEY,
    VERIFY_SETTING,
    os.environ.get("LOG_LEVEL", "INFO"),
)

if TARGET_URL and TARGET_URL.lower().startswith("http://"):
    logger.warning(
        "TARGET_URL uses http://; upstream may redirect to https:// and change request behavior: %s",
        TARGET_URL,
    )


@app.route("/health", methods=["GET"])
def health():
    logger.debug("healthcheck request from %s", request.remote_addr)
    return "OK", 200


@app.before_request
def before_request():
    g.request_start = time.time()


@app.after_request
def after_request(response):
    elapsed_ms = int((time.time() - g.request_start) * 1000)
    if request.path != "/health":
        logger.info(
            "request complete method=%s path=%s status=%s elapsed_ms=%s",
            request.method,
            request.path,
            response.status_code,
            elapsed_ms,
        )
    return response


@app.route(
    "/",
    defaults={"path": ""},
    methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"],
    provide_automatic_options=False,
)
@app.route(
    "/<path:path>",
    methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"],
    provide_automatic_options=False,
)
def proxy(path):
    request_path = f"/{path}" if path else "/"
    request_size = len(request.get_data(cache=True))
    logger.info(
        "incoming request method=%s path=%s query=%s remote=%s bytes=%s",
        request.method,
        request_path,
        request.query_string.decode("utf-8", "ignore"),
        request.remote_addr,
        request_size,
    )

    if not TARGET_URL:
        logger.error("TARGET_URL is not set; cannot proxy request")
        return Response("TARGET_URL is not set", status=500)

    try:
        url = f"{TARGET_URL.rstrip('/')}/{path}".rstrip("/")
        start_time = time.time()
        headers = {k: v for k, v in request.headers if k.lower() != "host"}
        headers["X-Forwarded-By"] = "mtls-bridge"

        logger.debug("forwarding request to upstream url=%s headers=%s", url, headers)

        resp = requests.request(
            method=request.method,
            url=url,
            headers=headers,
            data=request.get_data(cache=True),
            cookies=request.cookies,
            cert=(CLIENT_CERT, CLIENT_KEY),
            verify=VERIFY_SETTING,
            timeout=TIMEOUT,
        )

        elapsed_ms = int((time.time() - start_time) * 1000)
        logger.info(
            "upstream response status=%s url=%s elapsed_ms=%s response_bytes=%s",
            resp.status_code,
            url,
            elapsed_ms,
            len(resp.content),
        )

        excluded_headers = ["content-encoding", "content-length", "transfer-encoding", "connection"]
        response_headers = [
            (k, v)
            for k, v in resp.raw.headers.items()
            if k.lower() not in excluded_headers
        ]

        return Response(resp.content, resp.status_code, response_headers)

    except Exception as e:
        logger.exception("proxy request failed")
        return Response(str(e), status=500)


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)