Integrate OAuth authority and bounded Range fixes on current main

.github/workflows/integration-regressions.yml

@@ -29,6 +29,7 @@ on:
       - 'src/msg/storage/session.py'
       - 'src/msg/storage/read_only.py'
       - 'tests/test_oauth.py'
+      - 'tests/test_oauth_authority.py'
       - 'tests/test_http_effect_order.py'
       - 'tests/test_read_entry_contract_matrix.py'
       - 'tests/test_read_query_keyset_boundary.py'
@@ -65,6 +66,7 @@ on:
       - 'src/msg/storage/session.py'
       - 'src/msg/storage/read_only.py'
       - 'tests/test_oauth.py'
+      - 'tests/test_oauth_authority.py'
       - 'tests/test_http_effect_order.py'
       - 'tests/test_read_entry_contract_matrix.py'
       - 'tests/test_read_query_keyset_boundary.py'
@@ -182,7 +184,7 @@ jobs:
       - name: Run recovery, keyset, effect-order, path and webhook regressions
         run: >-
           python -m pytest -q
-          tests/test_oauth.py
+          tests/test_oauth.py tests/test_oauth_authority.py
           tests/test_http_effect_order.py tests/test_legacy_routes_removed.py
           tests/test_read_query_keyset_boundary.py tests/test_read_entry_contract_matrix.py
           tests/test_recovery_runtime_generation.py

src/msg/extensions/hosting.py

@@ -330,7 +330,7 @@ def hosted_range(request, size, etag):
     requested = request.headers.get('range')
     if not requested or request.headers.get('if-range') not in {None, etag}:
         return (0, size), 200, {}
-    match = re.fullmatch(r'bytes=(\d*)-(\d*)', requested)
+    match = re.fullmatch(r'bytes=(\d*)-(\d*)', requested) if len(requested) <= 128 else None
     require(match is not None and any(match.groups()), 'range_not_satisfiable')
     left, right = match.groups()
     if left:

src/msg/security/authentication.py

@@ -114,7 +114,9 @@ async def authenticate(self, request, session, *, entry):
             old = await session.credential(request.arguments['credential_id'])
             from msg.security.oauth import require_binding
 
-            await require_binding(session, old, now, self.oauth_config)
+            await require_binding(
+                session, old, now, self.oauth_config, custodial_ceiling=self.temporary_ceiling()
+            )
             require(
                 old.kind == 'token' and old.subject_id == request.subject, 'recovery_unavailable'
             )
@@ -193,7 +195,13 @@ async def authenticate(self, request, session, *, entry):
             ids = ()
             from msg.security.oauth import require_binding
 
-            await require_binding(session, credential, now, self.oauth_config)
+            await require_binding(
+                session,
+                credential,
+                now,
+                self.oauth_config,
+                custodial_ceiling=self.temporary_ceiling(),
+            )
         else:
             raise Failure('invalid_proof')
         if credential.revoked_at is not None:

src/msg/security/oauth.py

@@ -15,9 +15,10 @@
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
 
-from msg.core.codec import b64, canonical, loads, parse_time, wire
+from msg.core.codec import b64, canonical, decode, loads, parse_time, wire
 from msg.core.errors import Failure, require
-from msg.core.models import Credential
+from msg.core.models import CapabilityGrant, Credential
+from msg.security.policy import constraints_subset, scope_subset
 from msg.security.quarantine import require_live_authority
 
 DEVICE_GRANT = 'urn:ietf:params:oauth:grant-type:device_code'
@@ -67,7 +68,25 @@ def scopes_for(client, value):
     return scopes
 
 
-async def require_source(tx, body, now):
+async def require_ceiling(tx, ceiling, parent):
+    # A derived credential is a limit, never a durable copy of authority which
+    # its signing source has since lost. Reuse the same containment vocabulary
+    # as certificate/delegation validation, including scope and constraints.
+    for grant in ceiling:
+        require(
+            any([
+                allowed.capability == grant.capability
+                and allowed.version == grant.version
+                and grant.operations <= allowed.operations
+                and await scope_subset(grant.scope, allowed.scope, tx)
+                and constraints_subset(grant.constraints, allowed.constraints)
+                for allowed in parent
+            ]),
+            'invalid_grant',
+        )
+
+
+async def require_source(tx, body, now, *, custodial_ceiling):
     parent = await tx.credential(body['parent'])
     subject = await tx.subject(body['subject'])
     require(
@@ -85,19 +104,32 @@ async def require_source(tx, body, now):
             (subject.resource_id,),
         )
         require(subject.kind == 'custodial' and row == (parent.id, 'active'), 'invalid_grant')
+    # Vault keys deliberately carry no request grants. Custodial sessions retain
+    # their captured token limits under the current temporary policy, while the
+    # vault key above remains only their live identity/revocation binding.
+    await require_ceiling(
+        tx,
+        tuple(decode(CapabilityGrant, raw) for raw in body.get('ceiling', ())),
+        custodial_ceiling if body.get('custodial') else parent.ceiling,
+    )
     if body.get('session'):
         _, session = get(tx, body['session'], now)
         require(not session.get('revoked'), 'invalid_grant')
-        await require_source(tx, session, now)
+        await require_source(tx, session, now, custodial_ceiling=custodial_ceiling)
     return subject
 
 
-async def require_binding(tx, credential, now, config):
+async def require_binding(tx, credential, now, config, *, custodial_ceiling):
     api = tx.one('SELECT body FROM oauth_states WHERE id=?', ('api:' + credential.id,))
     if api is not None:
         source = loads(api[0])
         require(credential.source_credential_id == source['parent'], 'invalid_grant')
-        await require_source(tx, source, now)
+        await require_source(tx, source, now, custodial_ceiling=custodial_ceiling)
+        # API bindings predate OAuth families and store no captured ceiling.
+        # Their credential grants remain bounded by the current signing source.
+        await require_ceiling(
+            tx, credential.ceiling, (await tx.credential(source['parent'])).ceiling
+        )
     row = tx.one('SELECT body FROM oauth_states WHERE id=?', ('access:' + credential.id,))
     if row is None:
         require(
@@ -111,7 +143,10 @@ async def require_binding(tx, credential, now, config):
     require(credential.source_credential_id == family['parent'], 'invalid_grant')
     client = client_for(config, family['client_id'])
     require(set(family['scopes']) <= client.scopes and not family.get('revoked'), 'invalid_grant')
-    await require_source(tx, family, now)
+    await require_source(tx, family, now, custodial_ceiling=custodial_ceiling)
+    await require_ceiling(
+        tx, credential.ceiling, tuple(decode(CapabilityGrant, raw) for raw in family['ceiling'])
+    )
 
 
 class OAuthService:
@@ -241,7 +276,7 @@ async def pending_result(self, tx, value, client_id, kind):
             return None, 'authorization_pending'
         if body['status'] == 'denied':
             return None, 'access_denied'
-        await require_source(tx, body, now)
+        await require_source(tx, body, now, custodial_ceiling=self.app.temporary_ceiling())
         body['status'] = 'consumed'
         save(tx, locator['user'], body)
         if kind == 'login':
@@ -259,7 +294,9 @@ async def session(self, tx, cookie):
         id = state_id('session', cookie)
         expiry, body = get(tx, id, self.app.clock())
         require(not body.get('revoked'), 'invalid_grant')
-        await require_source(tx, body, self.app.clock())
+        await require_source(
+            tx, body, self.app.clock(), custodial_ceiling=self.app.temporary_ceiling()
+        )
         source = {
             name: body[name]
             for name in ('subject', 'parent', 'auth_version', 'custodial', 'ceiling', 'auth_time')
@@ -349,7 +386,7 @@ async def exchange(self, tx, args):
                 ),
                 'invalid_grant',
             )
-            await require_source(tx, body, now)
+            await require_source(tx, body, now, custodial_ceiling=self.app.temporary_ceiling())
             body['consumed'] = True
             body['family'] = 'family:' + uuid4().hex
             save(tx, id, body)
@@ -373,19 +410,18 @@ async def exchange(self, tx, args):
             require(not family.get('revoked'), 'invalid_grant')
             client = client_for(self.config, family['client_id'])
             require(set(family['scopes']) <= client.scopes, 'invalid_scope')
-            await require_source(tx, family, now)
+            await require_source(tx, family, now, custodial_ceiling=self.app.temporary_ceiling())
             old['used'] = True
             save(tx, id, old)
             # A refresh does not extend the family's absolute lifetime.
             return await self.tokens(tx, family, family_id=old['family'], expiry=expiry), None
         return None, 'unsupported_grant_type'
 
     async def tokens(self, tx, source, *, family_id=None, expiry=None):
-        from msg.core.codec import decode
-        from msg.core.models import CapabilityGrant
-
         now = self.app.clock()
-        subject = await require_source(tx, source, now)
+        subject = await require_source(
+            tx, source, now, custodial_ceiling=self.app.temporary_ceiling()
+        )
         if family_id is None:
             family_id = 'family:' + uuid4().hex
             expiry = now + timedelta(seconds=self.config.session_ttl)
@@ -496,7 +532,9 @@ async def bearer(self, tx, value):
             'invalid_token',
         )
         try:
-            await require_binding(tx, credential, now, self.config)
+            await require_binding(
+                tx, credential, now, self.config, custodial_ceiling=self.app.temporary_ceiling()
+            )
         except Failure as exc:
             if exc.code in {'recovery_runtime_stale', 'recovery_quarantined'}:
                 raise

src/msg/transports/http_routes.py

@@ -2431,7 +2431,11 @@ async def dispatch(request: Request):
                 status = 200
                 requested_range = request.headers.get('range')
                 if requested_range and (request.headers.get('if-range') in {None, etag}):
-                    match = re.fullmatch(r'bytes=(\d*)-(\d*)', requested_range)
+                    match = (
+                        re.fullmatch(r'bytes=(\d*)-(\d*)', requested_range)
+                        if len(requested_range) <= 128
+                        else None
+                    )
                     require(match is not None and any(match.groups()), 'range_not_satisfiable')
                     left, right = match.groups()
                     if left:

src/msg/transports/oauth_http.py

@@ -235,8 +235,11 @@ async def dispatch(self, request):
                 ],
             })
         # Rate limits commit independently, including rejected code guesses.
-        async with self.service.metadata.transaction(write=True) as tx:
-            self.oauth.rate(tx, request.client.host if request.client else '')
+        # Userinfo is a read: do not mutate retained OAuth state during GET or
+        # POST reads. Authentication still checks the current source/family.
+        if path != '/oauth/userinfo':
+            async with self.service.metadata.transaction(write=True) as tx:
+                self.oauth.rate(tx, request.client.host if request.client else '')
         if path == '/oauth/login':
             require(request.method == 'GET', 'method_not_allowed')
             async with self.service.metadata.transaction(write=True) as tx:
@@ -346,7 +349,7 @@ async def dispatch(self, request):
             'method_not_allowed',
         )
         args = await form(request) if request.method == 'POST' else {}
-        async with self.service.metadata.transaction(write=True) as tx:
+        async with self.service.metadata.transaction(write=path != '/oauth/userinfo') as tx:
             self.oauth.fence(tx)
             if path == '/oauth/device_authorization':
                 require(set(args) <= {'client_id', 'scope'}, 'invalid_request')

tests/test_hosting_same_origin.py

@@ -91,6 +91,15 @@ async def test_same_domain_hosted_html_head_304_range_raw_and_no_write_route(ins
         partial = await http.get('/@web-owner/web/index.html', headers={'Range': 'bytes=0-14'})
         assert partial.status_code == 206 and partial.content == html[:15]
         isolated(partial)
+        for malformed in (
+            'bytes=' + '9' * 5000 + '-',
+            'bytes=0-' + '9' * 5000,
+            'bytes=-' + '9' * 5000,
+        ):
+            refused = await http.get('/@web-owner/web/index.html', headers={'Range': malformed})
+            assert refused.status_code == 416
+            assert refused.json()['error'] == 'range_not_satisfiable'
+            isolated(refused)
         raw = await http.get(f'/_read/{published}/raw')
         assert raw.status_code == 200
         assert 'attachment' in raw.headers['content-disposition']

tests/test_http_range_bounds.py

@@ -0,0 +1,70 @@
+"""Malformed Range fields have bounded parsing and stable client errors."""
+
+import pytest
+from starlette.requests import Request
+
+from msg.core.errors import Failure
+from msg.extensions.hosting import hosted_range
+
+
+@pytest.mark.parametrize(
+    'value',
+    ['bytes=' + '9' * 5000 + '-', 'bytes=0-' + '9' * 5000, 'bytes=-' + '9' * 5000],
+    ids=['start', 'end', 'suffix'],
+)
+def test_hosted_range_rejects_oversized_numbers_with_contract_error(value):
+    request = Request({'type': 'http', 'headers': [(b'range', value.encode())]})
+    with pytest.raises(Failure, match='range_not_satisfiable'):
+        hosted_range(request, 12, '"etag"')
+
+
+@pytest.mark.parametrize(
+    'value,expected',
+    [('bytes=0-4', (0, 5)), ('bytes=5-', (5, 12)), ('bytes=-5', (7, 12)), ('bytes=-99', (0, 12))],
+)
+def test_hosted_range_keeps_normal_prefix_open_and_suffix_ranges(value, expected):
+    request = Request({'type': 'http', 'headers': [(b'range', value.encode())]})
+    interval, status, _ = hosted_range(request, 12, '"etag"')
+    assert interval == expected and status == 206
+
+
+def test_stale_if_range_ignores_range_before_parsing():
+    request = Request({
+        'type': 'http',
+        'headers': [(b'range', b'bytes=' + b'9' * 5000 + b'-'), (b'if-range', b'"stale"')],
+    })
+    assert hosted_range(request, 12, '"etag"') == ((0, 12), 200, {})
+
+
+@pytest.mark.asyncio
+async def test_raw_http_rejects_oversized_range_without_read_effects(installed, monkeypatch):
+    import httpx
+    from read_only_evidence import readonly_evidence
+    from test_service import call, register
+
+    from msg.transports.http import create_app
+
+    app, _ = installed
+    key, subject, _ = await register(app, 'range-reader')
+    created = await call(
+        app,
+        'content.post_create',
+        {'parent': '/main', 'body': 'range content'},
+        key=key,
+        subject=subject,
+    )
+    assert created.status == 'ok'
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=create_app(app)), base_url=app.settings.service_url
+    ) as http:
+        async with readonly_evidence(app, monkeypatch):
+            for value in (
+                'bytes=' + '9' * 5000 + '-',
+                'bytes=0-' + '9' * 5000,
+                'bytes=-' + '9' * 5000,
+            ):
+                result = await http.get(
+                    '/_id/' + created.resources[0].id + '/raw', headers={'Range': value}
+                )
+                assert result.status_code == 416
+                assert result.json()['error']['code'] == 'range_not_satisfiable'

tests/test_oauth.py

@@ -364,7 +364,7 @@ async def test_api_key_private_signature_rotation_delivery_and_header(oauth, tmp
 
 
 @pytest.mark.asyncio
-async def test_custodial_signup_login_and_device_consent_reuses_vault(oauth):
+async def test_custodial_signup_login_and_device_consent_reuses_vault(oauth, monkeypatch):
     app, _, _, http = oauth
     await http.get('/oauth/signup')
     binder = http.cookies.get('msg_login')
@@ -413,6 +413,21 @@ async def test_custodial_signup_login_and_device_consent_reuses_vault(oauth):
     assert response.status_code == 200, response.text
     async with app.metadata.transaction(write=False) as tx:
         assert tx.one('SELECT COUNT(*) FROM custodial_vault')[0] == 1
+        key_id = tx.one('SELECT signing_key_id FROM custodial_vault')[0]
+        assert (await tx.credential(key_id)).ceiling == ()
+    access = response.json()['access_token']
+    assert (
+        await http.get('/oauth/userinfo', headers={'Authorization': 'Bearer ' + access})
+    ).status_code == 200
+    # The vault never receives request grants to make login work. Current policy
+    # loss must invalidate already-issued custodial access without mutating reads.
+    from read_only_evidence import readonly_evidence
+
+    monkeypatch.setattr(app, 'temporary_ceiling', lambda: ())
+    monkeypatch.setattr(app.authenticator, 'temporary_ceiling', lambda: ())
+    async with readonly_evidence(app, monkeypatch):
+        denied = await http.get('/oauth/userinfo', headers={'Authorization': 'Bearer ' + access})
+        assert denied.status_code == 401 and denied.json()['error'] == 'invalid_token'
 
 
 @pytest.mark.parametrize(