"""MCP resource-server audience membership. Python 3.9+, standard library only. Claims are supplied already parsed. This does not decode a JWT, verify a signature, check expiry, call a network, or decide an HTTP status. Teaching bounds: audience strings 1..2048 characters, at most 32 audiences. """ import json import sys import unittest MAX_AUDIENCE_LENGTH = 2048 MAX_AUDIENCES = 32 def _audience_string(value, label): if type(value) is not str: raise ValueError(label + " must be a string") if value.strip() != value or not value or len(value) > MAX_AUDIENCE_LENGTH: raise ValueError(label + " must be 1..2048 characters without surrounding whitespace") if any(ord(ch) < 32 or ord(ch) == 127 for ch in value): raise ValueError(label + " contains a control character") if "#" in value: raise ValueError(label + " must not contain a fragment") return value def audience_values(aud): """Return presented audience strings. None means the claim is absent.""" if aud is None: return () if type(aud) is str: return (_audience_string(aud, "aud"),) if type(aud) is list: if not aud or len(aud) > MAX_AUDIENCES: raise ValueError("aud array must contain 1..32 strings") return tuple(_audience_string(item, "aud entry") for item in aud) raise ValueError("aud must be a string, an array of strings, or absent") def audience_decision(expected, aud, *, single_audience=False): """Exact membership of expected in aud. single_audience is local policy. It denies a match when aud names more than one distinct recipient. It does not replace a mismatch. """ if type(single_audience) is not bool: raise ValueError("single_audience must be a Boolean") expected = _audience_string(expected, "expected audience") values = audience_values(aud) distinct = len(set(values)) matched = expected in values if not values: reason = "audience_missing" allow = False elif not matched: reason = "audience_mismatch" allow = False elif single_audience and distinct != 1: reason = "single_audience_policy" allow = False else: reason = "audience_match" allow = True return { "allow": allow, "reason": reason, "audiences": list(values), } def decision_from_claims(expected, claims, *, single_audience=False): """Read aud and ignore every other claim, including iss and scope.""" if type(claims) is not dict: raise ValueError("claims must be a dict") aud = claims["aud"] if "aud" in claims else None return audience_decision(expected, aud, single_audience=single_audience) def demonstrate(): expected = "https://pay.example/mcp" samples = ( ("exact", {"aud": expected, "iss": "https://auth.example", "scope": "payments.read"}), ("other_server", {"aud": "https://docs.example/mcp", "iss": "https://auth.example"}), ("longer_path", {"aud": "https://pay.example/mcp-backup"}), ("trailing_slash", {"aud": "https://pay.example/mcp/"}), ("two_servers", {"aud": ["https://docs.example/mcp", expected]}), ("absent", {"iss": "https://auth.example", "scope": "payments.admin"}), ) cases = [] for name, claims in samples: row = {"case": name} for policy, flag in (("membership", False), ("single_audience", True)): decision = decision_from_claims(expected, claims, single_audience=flag) row[policy] = {"allow": decision["allow"], "reason": decision["reason"]} cases.append(row) return {"expected_audience": expected, "cases": cases} class AudienceTests(unittest.TestCase): expected = "https://pay.example/mcp" def test_exact_string_and_single_item_array(self): for aud in (self.expected, [self.expected]): decision = audience_decision(self.expected, aud) self.assertEqual(decision["allow"], True) self.assertEqual(decision["reason"], "audience_match") def test_other_server_is_denied(self): decision = audience_decision(self.expected, "https://docs.example/mcp") self.assertEqual(decision["reason"], "audience_mismatch") self.assertEqual(decision["allow"], False) def test_prefix_extensions_are_denied(self): for aud in ( "https://pay.example/mcp-backup", "https://pay.example/mcp/", "https://pay.example/mcp?x=1", "https://pay.example:443/mcp", "https://Pay.example/mcp", "https://p\u0430y.example/mcp", ): decision = audience_decision(self.expected, aud) self.assertEqual(decision["allow"], False, aud) self.assertEqual(decision["reason"], "audience_mismatch", aud) def test_shorter_prefix_is_denied(self): decision = audience_decision(self.expected, "https://pay.example") self.assertEqual(decision["allow"], False) def test_string_methods_that_must_not_be_used(self): longer = "https://pay.example/mcp-backup" shorter = "https://pay.example" self.assertTrue(longer.startswith(self.expected)) self.assertTrue(self.expected.startswith(shorter)) self.assertEqual(audience_decision(self.expected, longer)["allow"], False) self.assertEqual(audience_decision(self.expected, shorter)["allow"], False) def test_absent_claim_is_denied(self): decision = decision_from_claims(self.expected, {"iss": "https://auth.example"}) self.assertEqual(decision["reason"], "audience_missing") self.assertEqual(decision["allow"], False) self.assertEqual(audience_decision(self.expected, None)["reason"], "audience_missing") def test_other_claims_do_not_grant_access(self): claims = { "iss": "https://auth.example", "scope": "payments.admin", "aud": "https://docs.example/mcp", } decision = decision_from_claims(self.expected, claims) self.assertEqual(decision["allow"], False) self.assertEqual(set(decision), {"allow", "reason", "audiences"}) def test_multi_audience_policies_differ(self): aud = ["https://docs.example/mcp", self.expected] member = audience_decision(self.expected, aud, single_audience=False) single = audience_decision(self.expected, aud, single_audience=True) self.assertEqual(member["reason"], "audience_match") self.assertEqual(single["reason"], "single_audience_policy") self.assertEqual(single["allow"], False) def test_duplicate_of_one_audience_stays_single(self): aud = [self.expected, self.expected] decision = audience_decision(self.expected, aud, single_audience=True) self.assertEqual(decision["reason"], "audience_match") def test_multi_audience_mismatch_is_not_relabelled(self): decision = audience_decision( self.expected, ["https://docs.example/mcp", "https://other.example/mcp"], single_audience=True, ) self.assertEqual(decision["reason"], "audience_mismatch") def test_abstract_identifier_matches_exactly(self): decision = audience_decision("api://payments-mcp", "api://payments-mcp") self.assertEqual(decision["allow"], True) def test_invalid_inputs_raise(self): invalid_expected = ("", " https://pay.example/mcp", self.expected + "#frag", "bad\nvalue") for expected in invalid_expected: with self.assertRaises(ValueError): audience_decision(expected, self.expected) invalid_aud = (1, True, "", [], [""], ["ok", 2], ["https://pay.example/mcp#x"]) for aud in invalid_aud: with self.assertRaises(ValueError): audience_decision(self.expected, aud) with self.assertRaises(ValueError): audience_decision(self.expected, [self.expected] * (MAX_AUDIENCES + 1)) with self.assertRaises(ValueError): audience_decision(self.expected, "a" * (MAX_AUDIENCE_LENGTH + 1)) with self.assertRaises(ValueError): decision_from_claims(self.expected, ["aud"]) with self.assertRaises(ValueError): audience_decision(self.expected, self.expected, single_audience=1) def test_demonstration_rows(self): report = demonstrate() by_case = {row["case"]: row for row in report["cases"]} self.assertEqual(by_case["exact"]["membership"]["allow"], True) self.assertEqual(by_case["exact"]["single_audience"]["allow"], True) for name in ("other_server", "longer_path", "trailing_slash", "absent"): self.assertEqual(by_case[name]["membership"]["allow"], False) self.assertEqual(by_case[name]["single_audience"]["allow"], False) self.assertEqual(by_case["two_servers"]["membership"]["allow"], True) self.assertEqual(by_case["two_servers"]["single_audience"]["allow"], False) self.assertEqual(by_case["absent"]["membership"]["reason"], "audience_missing") self.assertEqual(by_case["two_servers"]["single_audience"]["reason"], "single_audience_policy") if __name__ == "__main__": if sys.argv[1:] == ["--test"]: unittest.main(argv=[sys.argv[0]]) elif sys.argv[1:]: raise SystemExit("Usage: python3 mcp-token-audience.py [--test]") else: print(json.dumps(demonstrate(), indent=2))