diff --git a/script/reverse_proxy/main.py b/script/reverse_proxy/main.py index 0956456..14febe8 100755 --- a/script/reverse_proxy/main.py +++ b/script/reverse_proxy/main.py @@ -25,7 +25,9 @@ la main. import argparse import asyncio import configparser +import ssl import sys +import time from dataclasses import dataclass # Une tête au-delà est refusée (431) : aucun navigateur n'en envoie de si @@ -58,14 +60,31 @@ FORWARDED = { RELAY_CHUNK = 64 * 1024 +# Délais de la mise en relation seulement, jamais du relais : une WebSocket +# ouverte reste muette des minutes entre deux notifications. +HEAD_TIMEOUT = 30.0 +CONNECT_TIMEOUT = 10.0 + @dataclass(frozen=True) class ProxyConfig: + """Réglages du mandataire. + + head_timeout borne l'attente de la tête d'une requête (408 au-delà), + connect_timeout la connexion à Odoo (504). trust_forwarded prolonge les + X-Forwarded-* reçus au lieu de les remplacer, pour un mandataire placé + derrière un autre. log reçoit une ligne par requête ; None le rend muet. + """ + odoo_host: str = "127.0.0.1" web_port: int = 8069 websocket_port: int = 8072 websocket_paths: tuple = DEFAULT_WEBSOCKET_PATHS forwarded_proto: str = "http" + head_timeout: float = HEAD_TIMEOUT + connect_timeout: float = CONNECT_TIMEOUT + trust_forwarded: bool = False + log: object = print def read_odoo_config(path): @@ -155,18 +174,38 @@ def rewrite_head(method, target, version, headers, client_ip, config): :return: la tête en octets, ligne vide finale comprise """ upgrade = is_upgrade(headers) - upgrade_value = next((v for n, v in headers if n.lower() == "upgrade"), "") - host = next((v for n, v in headers if n.lower() == "host"), "") + + def first(name): + return next((v for n, v in headers if n.lower() == name), "") + + upgrade_value = first("upgrade") + host = first("host") + forwarded_for = client_ip + real_ip = client_ip + forwarded_host = host + forwarded_proto = config.forwarded_proto + if config.trust_forwarded: + # Derrière un autre mandataire : sa chaîne est prolongée, et ce qu'il + # dit de l'hôte et du protocole du visiteur l'emporte. + chain = ", ".join( + v for n, v in headers if n.lower() == "x-forwarded-for" + ) + if chain: + forwarded_for = f"{chain}, {client_ip}" + real_ip = chain.split(",")[0].strip() + real_ip = first("x-real-ip") or real_ip + forwarded_host = first("x-forwarded-host") or host + forwarded_proto = first("x-forwarded-proto") or forwarded_proto kept = [ (n, v) for n, v in headers if n.lower() not in HOP_BY_HOP and n.lower() not in FORWARDED ] kept += [ - ("X-Forwarded-For", client_ip), - ("X-Real-IP", client_ip), - ("X-Forwarded-Host", host), - ("X-Forwarded-Proto", config.forwarded_proto), + ("X-Forwarded-For", forwarded_for), + ("X-Real-IP", real_ip), + ("X-Forwarded-Host", forwarded_host), + ("X-Forwarded-Proto", forwarded_proto), ] if upgrade: kept += [("Upgrade", upgrade_value), ("Connection", "Upgrade")] @@ -177,8 +216,8 @@ def rewrite_head(method, target, version, headers, client_ip, config): return ("\r\n".join(lines) + "\r\n\r\n").encode("latin-1") -async def _reply_error(writer, status, reason): - body = f"{status} {reason}\n".encode() +async def _reply_error(writer, status, reason, detail=""): + body = f"{status} {reason}\n{detail}".encode() writer.write( f"HTTP/1.1 {status} {reason}\r\nContent-Type: text/plain\r\n" f"Content-Length: {len(body)}\r\nConnection: close\r\n\r\n".encode() @@ -190,14 +229,19 @@ async def _reply_error(writer, status, reason): pass -async def _pipe(reader, writer, half_close): +async def _pipe(reader, writer, half_close, on_first=None): """Copie reader vers writer jusqu'à la fin du flux. half_close : à la fin, fermer seulement l'écriture (write_eof) plutôt que rien — le client qui a fini d'envoyer attend encore la réponse. + on_first reçoit le premier bloc lu, sans le retenir : le journal y lit + le statut de la réponse. """ try: while data := await reader.read(RELAY_CHUNK): + if on_first is not None: + on_first(data) + on_first = None writer.write(data) await writer.drain() if half_close and writer.can_write_eof(): @@ -206,44 +250,110 @@ async def _pipe(reader, writer, half_close): pass +def _status_of(chunk): + """Le statut d'une réponse d'après son premier bloc, « ? » sinon.""" + parts = chunk.split(b" ", 2) + if len(parts) >= 2 and parts[0].startswith(b"HTTP/"): + return parts[1].decode("latin-1") + return "?" + + +def unreachable_detail(config, port): + """Pourquoi Odoo ne répond pas sur ce port, en une ligne.""" + where = f"{config.odoo_host}:{port}" + if port == config.websocket_port: + return ( + f"Odoo ne répond pas sur {where} (bus). Le bus n'écoute que si" + " Odoo tourne avec workers >= 1.\n" + ) + return f"Odoo ne répond pas sur {where} (web). Odoo est-il démarré ?\n" + + +async def _open_upstream(host, port): + return await asyncio.open_connection(host, port) + + async def handle(reader, writer, config): - """Sert une connexion cliente : une requête, relayée puis fermée.""" + """Sert une connexion cliente : une requête, relayée puis fermée. + + Une ligne de journal par requête : client, méthode, cible, route (web + ou bus), statut et durée — celle d'une WebSocket va jusqu'à sa fermeture. + """ + started = time.monotonic() peer = writer.get_extra_info("peername") client_ip = peer[0] if peer else "" upstream_writer = None + request = "-" + route = "-" + status = {"code": "?"} + + def journal(): + if config.log is not None: + ms = int((time.monotonic() - started) * 1000) + config.log( + f"{client_ip} {request} → {route} {status['code']} {ms} ms" + ) + try: try: - head = await reader.readuntil(b"\r\n\r\n") + head = await asyncio.wait_for( + reader.readuntil(b"\r\n\r\n"), config.head_timeout + ) + except asyncio.TimeoutError: + await _reply_error(writer, 408, "Request Timeout") + return except asyncio.LimitOverrunError: + status["code"] = "431" await _reply_error(writer, 431, "Request Header Fields Too Large") + journal() return except asyncio.IncompleteReadError: return try: method, target, version, headers = parse_head(head) except ValueError: + status["code"] = "400" await _reply_error(writer, 400, "Bad Request") + journal() return - port = ( - config.websocket_port - if is_websocket_path(target, config) - else config.web_port - ) + request = f"{method} {target}" + bus = is_websocket_path(target, config) + route = "bus" if bus else "web" + port = config.websocket_port if bus else config.web_port try: - upstream_reader, upstream_writer = await asyncio.open_connection( - config.odoo_host, port + upstream_reader, upstream_writer = await asyncio.wait_for( + _open_upstream(config.odoo_host, port), config.connect_timeout ) - except OSError as e: - print(f"Odoo injoignable sur {config.odoo_host}:{port} : {e}") - await _reply_error(writer, 502, "Bad Gateway") + except asyncio.TimeoutError: + status["code"] = "504" + await _reply_error( + writer, + 504, + "Gateway Timeout", + unreachable_detail(config, port), + ) + journal() + return + except OSError: + status["code"] = "502" + await _reply_error( + writer, 502, "Bad Gateway", unreachable_detail(config, port) + ) + journal() return upstream_writer.write( rewrite_head(method, target, version, headers, client_ip, config) ) + + def note_status(chunk): + status["code"] = _status_of(chunk) + # Ce que le client a déjà envoyé après la tête — un début de corps — # est dans le tampon du lecteur : _pipe le relaie en premier. to_client = asyncio.create_task( - _pipe(upstream_reader, writer, half_close=False) + _pipe( + upstream_reader, writer, half_close=False, on_first=note_status + ) ) to_odoo = asyncio.create_task( _pipe(reader, upstream_writer, half_close=True) @@ -258,19 +368,67 @@ async def handle(reader, writer, config): await to_client for task in (to_client, to_odoo): task.cancel() + journal() finally: for w in (upstream_writer, writer): if w is not None: w.close() -async def serve(config, listen, port): - """Démarre l'écoute et rend le serveur asyncio, déjà à l'écoute.""" +async def serve(config, listen, port, ssl_context=None): + """Démarre l'écoute et rend le serveur asyncio, déjà à l'écoute. + + ssl_context : écoute en HTTPS ; Odoo, derrière, reste en HTTP clair. + """ return await asyncio.start_server( - lambda r, w: handle(r, w, config), listen, port, limit=MAX_HEAD + lambda r, w: handle(r, w, config), + listen, + port, + limit=MAX_HEAD, + ssl=ssl_context, ) +def make_ssl_context(cert, key): + """Contexte serveur TLS depuis un certificat et sa clé (PEM).""" + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(cert, key) + return context + + +async def probe(config): + """Tente une connexion aux deux ports d'Odoo. + + :return: {"web": bool, "bus": bool}, vrai quand le port accepte + """ + result = {} + for role, port in ( + ("web", config.web_port), + ("bus", config.websocket_port), + ): + try: + _, w = await asyncio.wait_for( + _open_upstream(config.odoo_host, port), config.connect_timeout + ) + w.close() + result[role] = True + except (OSError, asyncio.TimeoutError): + result[role] = False + return result + + +def startup_warnings(state, config): + """Les avertissements à dire au démarrage d'après probe(), [] sinon.""" + warnings = [] + if not state.get("web"): + warnings.append(unreachable_detail(config, config.web_port).strip()) + if not state.get("bus"): + warnings.append( + unreachable_detail(config, config.websocket_port).strip() + ) + return warnings + + def get_config(argv=None): parser = argparse.ArgumentParser( description=( @@ -303,11 +461,36 @@ def get_config(argv=None): ) parser.add_argument( "--forwarded-proto", - default="http", choices=("http", "https"), - help="Valeur de X-Forwarded-Proto, https derrière une terminaison TLS.", + help=( + "Valeur de X-Forwarded-Proto. Défaut : https avec --tls-cert," + " http sinon." + ), ) - return parser.parse_args(argv) + parser.add_argument( + "--tls-cert", + help="Certificat PEM : le mandataire écoute alors en HTTPS.", + ) + parser.add_argument("--tls-key", help="Clé PEM du certificat.") + parser.add_argument( + "--trust-forwarded", + action="store_true", + help=( + "Prolonger les X-Forwarded-* reçus au lieu de les remplacer, pour" + " un mandataire placé derrière un autre. À éviter face à des" + " navigateurs : ils pourraient se faire passer pour une autre" + " adresse." + ), + ) + parser.add_argument( + "--quiet", + action="store_true", + help="Ne pas écrire une ligne par requête.", + ) + args = parser.parse_args(argv) + if bool(args.tls_cert) != bool(args.tls_key): + parser.error("--tls-cert et --tls-key vont ensemble") + return args def config_from_args(args): @@ -316,25 +499,37 @@ def config_from_args(args): web_port=args.web_port, websocket_port=args.websocket_port, websocket_paths=tuple(args.websocket_path or DEFAULT_WEBSOCKET_PATHS), - forwarded_proto=args.forwarded_proto, + forwarded_proto=args.forwarded_proto + or ("https" if args.tls_cert else "http"), + trust_forwarded=args.trust_forwarded, + log=None if args.quiet else print, ) async def _run(args): config = config_from_args(args) - server = await serve(config, args.listen, args.port) + context = None + if args.tls_cert: + context = make_ssl_context(args.tls_cert, args.tls_key) + server = await serve(config, args.listen, args.port, ssl_context=context) + scheme = "https" if context else "http" print( - f"Mandataire sur {args.listen}:{args.port} → pages" + f"Mandataire sur {scheme}://{args.listen}:{args.port} → pages" f" {config.odoo_host}:{config.web_port}, bus" f" {config.odoo_host}:{config.websocket_port}" f" ({', '.join(config.websocket_paths)})." " Odoo doit tourner avec proxy_mode = True." ) + for warning in startup_warnings(await probe(config), config): + print(f"⚠️ {warning}") async with server: await server.serve_forever() def main(argv=None): + # Une ligne de journal doit paraître à sa requête, y compris quand la + # sortie est un tube ou un fichier, que Python tamponne sinon par blocs. + sys.stdout.reconfigure(line_buffering=True) try: asyncio.run(_run(get_config(argv))) except KeyboardInterrupt: diff --git a/test/test_reverse_proxy.py b/test/test_reverse_proxy.py index 4742f75..d39bd75 100644 --- a/test/test_reverse_proxy.py +++ b/test/test_reverse_proxy.py @@ -12,7 +12,12 @@ maquette de la bibliothèque réseau. import asyncio import gzip import hashlib +import os +import shutil +import ssl +import tempfile import unittest +from unittest.mock import patch from script.reverse_proxy import main as rp @@ -79,6 +84,8 @@ def reponse_fixe(corps=b"ok", extra=b""): class BaseProxy(unittest.IsolatedAsyncioTestCase): async def monter(self, web, ws, **options): + # Muet par défaut : le journal a ses propres tests. + options.setdefault("log", None) self.config = rp.ProxyConfig( odoo_host="127.0.0.1", web_port=web.port, @@ -307,6 +314,231 @@ class TestErreurs(BaseProxy): self.assertTrue(rep.startswith(b"HTTP/1.1 431"), rep[:40]) +class TestDelais(BaseProxy): + async def asyncSetUp(self): + self.web = await FauxOdoo(reponse_fixe()).demarrer() + self.ws = await FauxOdoo(reponse_fixe()).demarrer() + self.addAsyncCleanup(self.web.arreter) + self.addAsyncCleanup(self.ws.arreter) + + async def test_un_client_muet_recoit_408(self): + # Sans délai, une connexion ouverte sans requête garderait sa tâche + # pour toujours. + await self.monter(self.web, self.ws, head_timeout=0.2) + reader, writer = await asyncio.open_connection("127.0.0.1", self.port) + rep = await asyncio.wait_for(reader.read(), timeout=5) + writer.close() + self.assertTrue(rep.startswith(b"HTTP/1.1 408"), rep[:40]) + + async def test_un_odoo_qui_ne_repond_pas_a_la_connexion_rend_504(self): + async def jamais(host, port): + await asyncio.sleep(3600) + + await self.monter(self.web, self.ws, connect_timeout=0.2) + with patch.object(rp, "_open_upstream", jamais): + rep = await self.requete(b"GET /web HTTP/1.1\r\nHost: h\r\n\r\n") + self.assertTrue(rep.startswith(b"HTTP/1.1 504"), rep[:40]) + + async def test_une_websocket_n_est_pas_coupee_par_les_delais(self): + async def repondre(tete, reader, writer): + writer.write( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n" + b"Connection: Upgrade\r\n\r\n" + ) + await writer.drain() + while data := await reader.read(65536): + writer.write(data) + await writer.drain() + + ws = await FauxOdoo(repondre).demarrer() + self.addAsyncCleanup(ws.arreter) + await self.monter(self.web, ws, head_timeout=0.2, connect_timeout=0.2) + reader, writer = await asyncio.open_connection("127.0.0.1", self.port) + writer.write( + b"GET /websocket HTTP/1.1\r\nHost: h\r\nUpgrade: websocket\r\n" + b"Connection: Upgrade\r\n\r\n" + ) + await writer.drain() + await asyncio.wait_for(lire_tete(reader), timeout=5) + await asyncio.sleep(0.6) # plus long que les deux délais + writer.write(b"toujours") + await writer.drain() + self.assertEqual( + await asyncio.wait_for(reader.readexactly(8), timeout=5), + b"toujours", + ) + writer.close() + + +class TestJournal(BaseProxy): + async def asyncSetUp(self): + self.lignes = [] + self.web = await FauxOdoo(reponse_fixe()).demarrer() + self.ws = await FauxOdoo(reponse_fixe()).demarrer() + self.addAsyncCleanup(self.ws.arreter) + + async def test_une_ligne_par_requete_avec_route_et_statut(self): + self.addAsyncCleanup(self.web.arreter) + await self.monter(self.web, self.ws, log=self.lignes.append) + await self.requete(b"GET /web/login HTTP/1.1\r\nHost: h\r\n\r\n") + await asyncio.sleep(0.05) + self.assertEqual(len(self.lignes), 1, self.lignes) + ligne = self.lignes[0] + for attendu in ("GET /web/login", "web", "200", "ms"): + self.assertIn(attendu, ligne) + + async def test_un_echec_de_connexion_se_journalise(self): + await self.monter(self.web, self.ws, log=self.lignes.append) + await self.web.arreter() + await self.requete(b"GET /web HTTP/1.1\r\nHost: h\r\n\r\n") + await asyncio.sleep(0.05) + self.assertTrue(any("502" in ligne for ligne in self.lignes)) + + async def test_quiet_ne_journalise_rien(self): + self.addAsyncCleanup(self.web.arreter) + await self.monter(self.web, self.ws, log=None) + rep = await self.requete(b"GET /web HTTP/1.1\r\nHost: h\r\n\r\n") + self.assertTrue(rep.startswith(b"HTTP/1.1 200")) + + +class TestUn502QuiExplique(BaseProxy): + async def asyncSetUp(self): + self.web = await FauxOdoo(reponse_fixe()).demarrer() + self.ws = await FauxOdoo(reponse_fixe()).demarrer() + + async def test_le_bus_ferme_nomme_le_port_et_workers(self): + self.addAsyncCleanup(self.web.arreter) + await self.monter(self.web, self.ws) + port = self.ws.port + await self.ws.arreter() + rep = await self.requete(b"GET /websocket HTTP/1.1\r\nHost: h\r\n\r\n") + self.assertTrue(rep.startswith(b"HTTP/1.1 502")) + self.assertIn(f":{port}".encode(), rep) + self.assertIn(b"workers", rep) + + async def test_le_web_ferme_nomme_son_port(self): + self.addAsyncCleanup(self.ws.arreter) + await self.monter(self.web, self.ws) + port = self.web.port + await self.web.arreter() + rep = await self.requete(b"GET /web HTTP/1.1\r\nHost: h\r\n\r\n") + self.assertIn(f":{port}".encode(), rep) + self.assertNotIn(b"workers", rep) + + +class TestSondeAuDemarrage(unittest.IsolatedAsyncioTestCase): + async def test_un_bus_ferme_se_signale_des_le_demarrage(self): + web = await FauxOdoo(reponse_fixe()).demarrer() + self.addAsyncCleanup(web.arreter) + ferme = await FauxOdoo(reponse_fixe()).demarrer() + port_ferme = ferme.port + await ferme.arreter() + config = rp.ProxyConfig(web_port=web.port, websocket_port=port_ferme) + etat = await rp.probe(config) + self.assertEqual(etat, {"web": True, "bus": False}) + avis = " ".join(rp.startup_warnings(etat, config)) + self.assertIn(str(port_ferme), avis) + self.assertIn("workers", avis) + + async def test_rien_a_dire_quand_tout_repond(self): + web = await FauxOdoo(reponse_fixe()).demarrer() + ws = await FauxOdoo(reponse_fixe()).demarrer() + self.addAsyncCleanup(web.arreter) + self.addAsyncCleanup(ws.arreter) + config = rp.ProxyConfig(web_port=web.port, websocket_port=ws.port) + self.assertEqual( + rp.startup_warnings(await rp.probe(config), config), [] + ) + + +class TestConfianceEnAmont(BaseProxy): + async def asyncSetUp(self): + self.web = await FauxOdoo(reponse_fixe()).demarrer() + self.ws = await FauxOdoo(reponse_fixe()).demarrer() + self.addAsyncCleanup(self.web.arreter) + self.addAsyncCleanup(self.ws.arreter) + + async def test_la_chaine_est_prolongee_et_non_remplacee(self): + await self.monter(self.web, self.ws, trust_forwarded=True) + await self.requete( + b"GET /web HTTP/1.1\r\nHost: interne\r\n" + b"X-Forwarded-For: 203.0.113.9\r\nX-Forwarded-Proto: https\r\n" + b"X-Forwarded-Host: public.test\r\n\r\n" + ) + recus = entetes(self.web.tetes[0]) + self.assertEqual( + [v for n, v in recus if n == "x-forwarded-for"], + ["203.0.113.9, 127.0.0.1"], + ) + self.assertEqual( + [v for n, v in recus if n == "x-forwarded-proto"], ["https"] + ) + self.assertEqual( + [v for n, v in recus if n == "x-forwarded-host"], ["public.test"] + ) + self.assertEqual( + [v for n, v in recus if n == "x-real-ip"], ["203.0.113.9"] + ) + + +@unittest.skipUnless(shutil.which("openssl"), "openssl absent") +class TestTLS(BaseProxy): + async def test_une_requete_https_arrive_a_odoo_marquee_https(self): + from script.reverse_proxy import local_cert + + dossier = os.path.join(tempfile.mkdtemp(), "tls") + self.addCleanup(shutil.rmtree, os.path.dirname(dossier)) + chemins = local_cert.issue(dossier, ["localhost", "127.0.0.1"]) + web = await FauxOdoo(reponse_fixe(b"secret")).demarrer() + ws = await FauxOdoo(reponse_fixe()).demarrer() + self.addAsyncCleanup(web.arreter) + self.addAsyncCleanup(ws.arreter) + args = rp.get_config( + [ + "--tls-cert", + chemins["server_crt"], + "--tls-key", + chemins["server_key"], + ] + ) + self.config = rp.config_from_args(args) + self.assertEqual(self.config.forwarded_proto, "https") + self.config = rp.ProxyConfig( + web_port=web.port, + websocket_port=ws.port, + forwarded_proto=self.config.forwarded_proto, + ) + serveur = await rp.serve( + self.config, + "127.0.0.1", + 0, + ssl_context=rp.make_ssl_context( + chemins["server_crt"], chemins["server_key"] + ), + ) + self.addAsyncCleanup(self._arreter, serveur) + port = serveur.sockets[0].getsockname()[1] + client = ssl.create_default_context(cafile=chemins["ca_crt"]) + reader, writer = await asyncio.open_connection( + "127.0.0.1", port, ssl=client + ) + writer.write(b"GET /web HTTP/1.1\r\nHost: localhost\r\n\r\n") + await writer.drain() + rep = await asyncio.wait_for(reader.read(), timeout=10) + writer.close() + self.assertTrue(rep.endswith(b"secret"), rep[-40:]) + recus = dict(entetes(web.tetes[0])) + self.assertEqual(recus["x-forwarded-proto"], "https") + + async def _arreter(self, serveur): + serveur.close() + await serveur.wait_closed() + + def test_un_certificat_sans_cle_est_refuse(self): + with self.assertRaises(SystemExit): + rp.get_config(["--tls-cert", "/tmp/x.crt"]) + + class TestLectureDeConfig(unittest.TestCase): """Les ports et réglages lus dans un config.conf d'Odoo."""