fix(plugins): reject plaintext feed redirect hops

hermes_cli/plugins_updates.py

@@ -367,9 +367,16 @@ def default_fetch(url: str) -> str:
     """
     import urllib.request
 
-    with urllib.request.urlopen(https_update_url(url), timeout=_FETCH_TIMEOUT) as resp:
-        # urllib follows redirects automatically; do not trust a feed moved onto plaintext.
-        https_update_url(resp.geturl())
+    class HTTPSFeedRedirectHandler(urllib.request.HTTPRedirectHandler):
+        def redirect_request(self, req, fp, code, msg, headers, newurl):
+            redirected = super().redirect_request(req, fp, code, msg, headers, newurl)
+            if redirected is not None and not redirected.full_url.lower().startswith("https://"):
+                fp.close()
+                raise ValueError("update_url redirect must use https://")
+            return redirected
+
+    opener = urllib.request.build_opener(HTTPSFeedRedirectHandler())
+    with opener.open(https_update_url(url), timeout=_FETCH_TIMEOUT) as resp:
         data = resp.read(_MAX_FEED_BYTES)
     return data.decode("utf-8", errors="replace")
 

tests/hermes_cli/test_plugins_check_updates.py

@@ -484,23 +484,66 @@ def test_default_fetch_refuses_non_https_feeds_before_any_request(monkeypatch, u
         default_fetch(url)
 
 
-def test_default_fetch_refuses_https_to_http_redirect_before_reading(monkeypatch):
+@pytest.fixture
+def feed_redirect_server(monkeypatch):
+    """Exercise urllib's redirect machinery without requiring a TLS certificate."""
+    import http.client
+    import threading
     import urllib.request
-    from hermes_cli.plugins_updates import default_fetch
+    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
-    class RedirectedResponse:
-        def __enter__(self):
-            return self
+    visited = []
 
-        def __exit__(self, *args):
+    class FeedHandler(BaseHTTPRequestHandler):
+        def do_GET(self):
+            visited.append(self.path)
+            if self.path == "/start":
+                target = f"http://127.0.0.1:{self.server.server_port}/middle"
+            elif self.path == "/middle":
+                target = f"https://127.0.0.1:{self.server.server_port}/feed"
+            elif self.path == "/secure":
+                target = f"https://127.0.0.1:{self.server.server_port}/feed"
+            else:
+                self.send_response(200)
+                self.end_headers()
+                self.wfile.write(b"version: 1.2.0\n")
+                return
+            self.send_response(302)
+            self.send_header("Location", target)
+            self.end_headers()
+
+        def log_message(self, *_args):
             pass
 
-        def geturl(self):
-            return "http://feed.example/insecure.yml"
+    # Only the HTTPS transport is replaced; the redirect handler and HTTP
+    # transport stay real, so the server can observe a forbidden HTTP hop.
+    monkeypatch.setattr(
+        urllib.request.HTTPSHandler, "https_open",
+        lambda self, req: self.do_open(http.client.HTTPConnection, req),
+    )
+    server = ThreadingHTTPServer(("127.0.0.1", 0), FeedHandler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"https://127.0.0.1:{server.server_port}", visited
+    finally:
+        server.shutdown()
+        thread.join()
+        server.server_close()
 
-        def read(self, *_):
-            raise AssertionError("insecure feed must not be read")
 
-    monkeypatch.setattr(urllib.request, "urlopen", lambda *_a, **_kw: RedirectedResponse())
+def test_default_fetch_refuses_intermediate_plaintext_redirect(feed_redirect_server):
+    from hermes_cli.plugins_updates import default_fetch
+
+    base, visited = feed_redirect_server
     with pytest.raises(ValueError, match="https://"):
-        default_fetch("https://feed.example/secure.yml")
+        default_fetch(base + "/start")
+    assert visited == ["/start"]
+
+
+def test_default_fetch_follows_https_redirect(feed_redirect_server):
+    from hermes_cli.plugins_updates import default_fetch
+
+    base, visited = feed_redirect_server
+    assert default_fetch(base + "/secure") == "version: 1.2.0\n"
+    assert visited == ["/secure", "/feed"]