Reject credential-bearing redirects

CHANGES.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.1.1
+
+- Reject HTTP redirects so API credentials cannot be forwarded to another origin.
+- Add an integration test proving a redirect target receives no request.
+- Attribute GitHub installs accurately instead of implying Exchange publication.
+
 ## 0.1.0
 
 - Add an explicit text-to-speech action that writes private WAV output.

actions/lib/brainiall_action.py

@@ -16,13 +16,23 @@
 import socket
 import tempfile
 from urllib.error import HTTPError, URLError
-from urllib.request import Request, urlopen
+from urllib.request import HTTPRedirectHandler, Request, build_opener
 
 from st2common.runners.base_action import Action
 
 
 API_BASE_URL = "https://api.brainiall.com"
-SOURCE = "stackstorm-exchange-c9"
+SOURCE = "stackstorm-github-c9"
+
+
+class _NoRedirectHandler(HTTPRedirectHandler):
+    """Fail closed instead of forwarding credentials to a redirect target."""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        return None
+
+
+NO_REDIRECT_OPENER = build_opener(_NoRedirectHandler())
 
 
 class BrainiallRequestError(RuntimeError):
@@ -57,7 +67,7 @@ def _post(self, path, payload, content_type, timeout, max_response_bytes):
         )
 
         try:
-            response = urlopen(request, timeout=timeout)
+            response = NO_REDIRECT_OPENER.open(request, timeout=timeout)
         except HTTPError as exc:
             exc.close()
             raise BrainiallRequestError(

pack.yaml

@@ -8,7 +8,7 @@ keywords:
   - tts
   - pdf
   - markdown
-version: 0.1.0
+version: 0.1.1
 python_versions:
   - "3"
 author: BRAINIALL Team

scripts/validate_pack.py

@@ -41,6 +41,7 @@ def validate_files():
         "actions/pdf_to_markdown.yaml",
         "tests/test_text_to_speech.py",
         "tests/test_pdf_to_markdown.py",
+        "tests/test_redirect_policy.py",
     ]
     missing = [path for path in required if not (ROOT / path).is_file()]
     if missing:

tests/test_pdf_to_markdown.py

@@ -43,12 +43,12 @@ def test_sends_bounded_multipart_and_writes_private_markdown(self):
             action = PdfToMarkdownAction({"api_key": TEST_KEY})
 
             with mock.patch(
-                "lib.brainiall_action.urlopen",
+                "lib.brainiall_action.NO_REDIRECT_OPENER.open",
                 return_value=FakeResponse(MARKDOWN_BYTES),
-            ) as urlopen_mock:
+            ) as open_mock:
                 result = action.run(pdf_path, output_path, "1-2", "markdown")
 
-            request = urlopen_mock.call_args[0][0]
+            request = open_mock.call_args[0][0]
             headers = lower_headers(request)
             self.assertEqual(
                 request.full_url,
@@ -77,23 +77,23 @@ def test_rejects_bad_page_range_and_symlink_before_network(self):
             os.symlink(pdf_path, symlink_path)
             action = PdfToMarkdownAction({"api_key": TEST_KEY})
 
-            with mock.patch("lib.brainiall_action.urlopen") as urlopen_mock:
+            with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open") as open_mock:
                 with self.assertRaises(ValueError):
                     action.run(pdf_path, output_path, "3-1")
                 with self.assertRaises(ValueError):
                     action.run(symlink_path, output_path)
-            urlopen_mock.assert_not_called()
+            open_mock.assert_not_called()
 
     def test_rejects_output_extension_before_metered_request(self):
         with tempfile.TemporaryDirectory() as directory:
             pdf_path = os.path.join(directory, "input.pdf")
             output_path = os.path.join(directory, "output.txt")
             self._write(pdf_path, PDF_BYTES)
             action = PdfToMarkdownAction({"api_key": TEST_KEY})
-            with mock.patch("lib.brainiall_action.urlopen") as urlopen_mock:
+            with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open") as open_mock:
                 with self.assertRaises(ValueError):
                     action.run(pdf_path, output_path)
-            urlopen_mock.assert_not_called()
+            open_mock.assert_not_called()
 
     def test_http_error_does_not_echo_key_or_replace_output(self):
         with tempfile.TemporaryDirectory() as directory:
@@ -110,7 +110,7 @@ def test_http_error_does_not_echo_key_or_replace_output(self):
             )
             action = PdfToMarkdownAction({"api_key": TEST_KEY})
 
-            with mock.patch("lib.brainiall_action.urlopen", side_effect=error):
+            with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open", side_effect=error):
                 with self.assertRaises(BrainiallRequestError) as raised:
                     action.run(pdf_path, output_path)
 

tests/test_redirect_policy.py

@@ -0,0 +1,97 @@
+# Copyright 2026 BRAINIALL
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# 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.
+
+import contextlib
+import http.server
+import socketserver
+import threading
+import unittest
+
+from common import install_st2_stub
+
+
+install_st2_stub()
+
+from lib.brainiall_action import BrainiallAction, BrainiallRequestError  # noqa: E402
+
+
+TEST_KEY = "redirect-policy-test-key"
+
+
+class _RedirectSource(http.server.BaseHTTPRequestHandler):
+    target_url = None
+
+    def do_POST(self):
+        self.send_response(302)
+        self.send_header("Location", self.target_url)
+        self.end_headers()
+
+    def log_message(self, format_string, *args):
+        pass
+
+
+class _RedirectTarget(http.server.BaseHTTPRequestHandler):
+    requests = []
+
+    def do_GET(self):
+        self.requests.append(dict(self.headers.items()))
+        self.send_response(200)
+        self.end_headers()
+        self.wfile.write(b"unexpected")
+
+    def do_POST(self):
+        self.requests.append(dict(self.headers.items()))
+        self.send_response(200)
+        self.end_headers()
+        self.wfile.write(b"unexpected")
+
+    def log_message(self, format_string, *args):
+        pass
+
+
+@contextlib.contextmanager
+def _server(handler):
+    instance = socketserver.TCPServer(("127.0.0.1", 0), handler)
+    thread = threading.Thread(target=instance.serve_forever)
+    thread.daemon = True
+    thread.start()
+    try:
+        yield instance
+    finally:
+        instance.shutdown()
+        instance.server_close()
+        thread.join()
+
+
+class RedirectPolicyTest(unittest.TestCase):
+    def test_redirect_is_rejected_without_contacting_target(self):
+        _RedirectTarget.requests = []
+        with _server(_RedirectTarget) as target:
+            _RedirectSource.target_url = "http://127.0.0.1:{0}/capture".format(
+                target.server_address[1]
+            )
+            with _server(_RedirectSource) as source:
+                action = BrainiallAction({"api_key": TEST_KEY})
+                source_url = "http://127.0.0.1:{0}".format(source.server_address[1])
+
+                with unittest.mock.patch("lib.brainiall_action.API_BASE_URL", source_url):
+                    with self.assertRaises(BrainiallRequestError) as raised:
+                        action._post("/redirect", b"{}", "application/json", 3, 100)
+
+        self.assertEqual(str(raised.exception), "Brainiall API request failed with HTTP 302.")
+        self.assertEqual(_RedirectTarget.requests, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_text_to_speech.py

@@ -42,19 +42,19 @@ def test_writes_private_wav_and_returns_only_metadata(self):
             action = TextToSpeechAction({"api_key": TEST_KEY})
 
             with mock.patch(
-                "lib.brainiall_action.urlopen",
+                "lib.brainiall_action.NO_REDIRECT_OPENER.open",
                 return_value=FakeResponse(WAV_BYTES),
-            ) as urlopen_mock:
+            ) as open_mock:
                 result = action.run(TEST_TEXT, output_path, "af_heart", 1.25)
 
-            request = urlopen_mock.call_args[0][0]
+            request = open_mock.call_args[0][0]
             headers = lower_headers(request)
             request_json = json.loads(request.data.decode("utf-8"))
 
             self.assertEqual(request.full_url, "https://api.brainiall.com/v1/tts/synthesize")
             self.assertEqual(headers["authorization"], "Bearer " + TEST_KEY)
             self.assertEqual(headers["ocp-apim-subscription-key"], TEST_KEY)
-            self.assertEqual(headers["x-brainiall-source"], "stackstorm-exchange-c9")
+            self.assertEqual(headers["x-brainiall-source"], "stackstorm-github-c9")
             self.assertEqual(request_json["text"], TEST_TEXT)
             self.assertNotIn(TEST_KEY.encode("utf-8"), request.data)
             self.assertNotIn(TEST_KEY, request.full_url)
@@ -78,7 +78,7 @@ def test_http_error_does_not_echo_key_content_or_replace_output(self):
             )
             action = TextToSpeechAction({"api_key": TEST_KEY})
 
-            with mock.patch("lib.brainiall_action.urlopen", side_effect=error):
+            with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open", side_effect=error):
                 with self.assertRaises(BrainiallRequestError) as raised:
                     action.run(TEST_TEXT, output_path)
 
@@ -91,19 +91,19 @@ def test_rejects_invalid_voice_before_network(self):
         with tempfile.TemporaryDirectory() as directory:
             output_path = os.path.join(directory, "speech.wav")
             action = TextToSpeechAction({"api_key": TEST_KEY})
-            with mock.patch("lib.brainiall_action.urlopen") as urlopen_mock:
+            with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open") as open_mock:
                 with self.assertRaises(ValueError):
                     action.run(TEST_TEXT, output_path, "voice\nheader")
-            urlopen_mock.assert_not_called()
+            open_mock.assert_not_called()
 
     def test_rejects_invalid_output_path_before_metered_request(self):
         with tempfile.TemporaryDirectory() as directory:
             output_path = os.path.join(directory, "speech.txt")
             action = TextToSpeechAction({"api_key": TEST_KEY})
-            with mock.patch("lib.brainiall_action.urlopen") as urlopen_mock:
+            with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open") as open_mock:
                 with self.assertRaises(ValueError):
                     action.run(TEST_TEXT, output_path)
-            urlopen_mock.assert_not_called()
+            open_mock.assert_not_called()
 
 
 class PathReader(object):