Bind password setup grants to approved company welcome pages

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c

identity-provisioner/tests/test_company_return_http.py

@@ -0,0 +1,77 @@
+import json
+import pathlib
+import sys
+import threading
+import unittest
+from http.server import ThreadingHTTPServer
+from urllib.error import HTTPError
+from urllib.parse import urlencode, urlsplit, parse_qs
+from urllib.request import Request, urlopen
+
+sys.path.insert(0, str(pathlib.Path(__file__).parents[1]))
+from password_setup import PasswordSetupGrants
+from provisioner import Result
+from server import Handler
+
+
+class CompanyReturnHTTPTests(unittest.TestCase):
+    def setUp(self):
+        self.calls = []
+        class Provisioner:
+            def provision(self, payload):
+                return Result('fixture', 'uid=recipient,ou=people', 'password_setup_required', False)
+        class TestHandler(Handler):
+            service_token = 'fixture-only-service-token'
+            provisioner = Provisioner()
+            password_setups = PasswordSetupGrants(
+                public_url='https://kc.example/setup/password',
+                setter=lambda subject, password: self.calls.append(subject),
+                tenant_returns={'tenant:trial:demo': 'https://app.example/demo/'},
+            )
+        self.handler = TestHandler
+        self.server = ThreadingHTTPServer(('127.0.0.1', 0), TestHandler)
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        self.base = 'http://127.0.0.1:' + str(self.server.server_port)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join()
+
+    def issue(self):
+        payload = {'user_id': 'fixture', 'tenant': 'tenant:trial:demo',
+                   'primary_email': 'fixture@example.test', 'idempotency_key': 'fixture-long-idempotency',
+                   'correlation_id': 'fixture', 'return_to': 'https://evil.example/'}
+        request = Request(self.base + '/v1/identities/provision', data=json.dumps(payload).encode(),
+                          headers={'Authorization': 'Bearer fixture-only-service-token'})
+        with urlopen(request) as response:
+            url = json.load(response)['password_setup_url']
+        return parse_qs(urlsplit(url).query)['token'][0]
+
+    def submit(self, token):
+        request = Request(self.base + '/setup/password', data=urlencode({
+            'token': token, 'password': 'fixture-long-password', 'confirmation': 'fixture-long-password',
+            'return_to': 'https://evil.example/',
+        }).encode(), headers={'Content-Type': 'application/x-www-form-urlencoded'})
+        try:
+            response = urlopen(request)
+        except HTTPError as error:
+            response = error
+        with response:
+            return response.status, response.read().decode(), response.headers
+
+    def test_authenticated_issue_and_completion_ignore_browser_return(self):
+        token = self.issue()
+        status, body, headers = self.submit(token)
+        self.assertEqual(200, status)
+        self.assertIn('https://app.example/demo/', body)
+        self.assertNotIn('evil.example', body)
+        self.assertNotIn(token, body)
+        self.assertNotIn('recipient', body)
+        self.assertEqual('no-referrer', headers['Referrer-Policy'])
+        self.assertEqual(['recipient'], self.calls)
+        status, body, headers = self.submit(token)
+        self.assertEqual(400, status)
+        self.assertNotIn('https://app.example/demo/', body)
+        self.assertEqual(['recipient'], self.calls)

identity-provisioner/tests/test_password_setup.py

@@ -47,3 +47,38 @@ class PasswordSetupGrantTests(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "12 characters"):
             self.grants.consume(token, "too-short")
         self.assertTrue(self.grants.valid(token))
+
+
+class CompanyReturnTests(unittest.TestCase):
+    def test_return_is_bound_to_grant_and_cannot_be_changed_by_browser(self):
+        mapping = {"tenant:trial:demo-company": "https://vergabe.example/demo-company/"}
+        grants = PasswordSetupGrants(public_url="https://kc.example/setup/password",
+                                     setter=lambda *args: None, tenant_returns=mapping)
+        url = grants.issue("recipient", "tenant:trial:demo-company")
+        token = url.partition("token=")[2]
+        mapping["tenant:trial:demo-company"] = "https://attacker.example/"
+        self.assertNotIn("recipient", url)
+        self.assertNotIn("return", url)
+        self.assertEqual("https://vergabe.example/demo-company/", grants.consume(token, "test-password-long"))
+        with self.assertRaises(ValueError):
+            grants.consume(token, "test-password-long")
+
+    def test_unknown_tenant_has_no_return_and_bad_targets_fail(self):
+        for target in ("http://example.test/", "https://example.test/?next=evil",
+                       "https://example.test/#fragment", "https://user:password@example.test/"):
+            with self.assertRaises(ValueError):
+                PasswordSetupGrants(public_url="https://kc.example/setup/password",
+                                    setter=lambda *args: None, tenant_returns={"tenant:trial:demo": target})
+        grants = PasswordSetupGrants(public_url="https://kc.example/setup/password", setter=lambda *args: None)
+        token = grants.issue("recipient", "tenant:unknown").partition("token=")[2]
+        self.assertEqual("", grants.consume(token, "test-password-long"))
+
+    def test_expiry_and_failure_never_release_a_return(self):
+        clock = [0]
+        grants = PasswordSetupGrants(public_url="https://kc.example/setup/password",
+                                    setter=lambda *args: None, clock=lambda: clock[0], ttl_seconds=1,
+                                    tenant_returns={"tenant:trial:demo": "https://app.example/demo/"})
+        token = grants.issue("recipient", "tenant:trial:demo").partition("token=")[2]
+        clock[0] = 2
+        with self.assertRaises(ValueError):
+            grants.consume(token, "test-password-long")