feat: read-only /@user/receipts projection of stored OperationResults

src/msg/plugins/communication.py

@@ -897,6 +897,8 @@ async def can_read(rid):
     install_following(app, op)
     from msg.plugins.watches import install as install_watches
     install_watches(app, op)
+    from msg.plugins.receipts import install as install_receipts
+    install_receipts(app, op)
     finish((ResourceTypeSpec(name='watch',version=1,container=False,content_schema=None,
                              operations=frozenset(),relations=frozenset()),
             ResourceTypeSpec(name='claim',version=1,container=False,content_schema=None,

src/msg/plugins/receipts.py

@@ -0,0 +1,76 @@
+"""A private, read-only projection of the caller's persisted OperationResults.
+
+Receipts are the signed results the executor already committed; this view adds
+no receipt state, table or writable resource. Stored bytes are never rewritten.
+"""
+from msg.core.codec import decode, digest, loads
+from msg.core.errors import Failure, require
+from msg.core.models import HandlerOutput, ResourceRef
+from msg.plugins.common import operation_id
+from msg.plugins.discovery import next_link, visible
+from msg.plugins.schemas import STRING, obj
+
+REQUEST_ID = {'type': 'string', 'minLength': 1, 'maxLength': 128}
+
+
+async def _subject(app, ctx, request, tx):
+    subject = ctx.principal.subject
+    require(subject is not None, 'authentication_required')
+    await app.authorizer.require_base(ctx.principal, operation_id(request), subject, tx)
+    return subject
+
+
+async def _project(app, ctx, request, tx, raw):
+    saved = loads(raw)
+    resources = []
+    # A receipt names resources as they were at commit time; access may have
+    # been revoked since, so only currently readable references are disclosed.
+    for value in saved.get('resources') or ():
+        ref = decode(ResourceRef, value)
+        try:
+            shown = await visible(app, ctx, request, tx, ref.id)
+        except Failure as exc:
+            if exc.code != 'not_found':
+                raise
+            shown = False
+        if shown:
+            resources.append(value)
+    # data/output may carry delivered credentials or other presentation-only
+    # material, so they are deliberately not part of this projection.
+    item = {'request_id': saved['request_id'], 'operation': saved['operation'],
+            'status': saved['status'], 'committed_at': saved.get('committed_at'),
+            'resources': resources, 'signed': saved.get('receipt') is not None}
+    error = saved.get('error')
+    if error:
+        item['error_code'] = error['code']
+    return item
+
+
+def install(app, op):
+    @op('communication.receipt_list', obj({'limit': {'type': 'integer', 'minimum': 1, 'maximum': 200},
+                                           'cursor': STRING}), effect='read')
+    async def receipt_list(ctx, request, tx):
+        subject = await _subject(app, ctx, request, tx)
+        binding = digest({'subject': subject})
+        cursor = request.arguments.get('cursor')
+        after = app.cursors.decode(cursor, request.operation, binding) if cursor else ''
+        require(type(after) is str, 'invalid_cursor')
+        limit = request.arguments.get('limit', 50)
+        rows = tx.rows('SELECT request_id,body FROM results WHERE subject=? AND request_id>? '
+                       'ORDER BY request_id LIMIT ?', (subject, after, limit + 1))
+        items = [await _project(app, ctx, request, tx, raw) for _, raw in rows[:limit]]
+        data = {'items': items}
+        if len(rows) > limit:
+            token = app.cursors.encode(request.operation, binding, rows[limit - 1][0])
+            data.update(cursor=token,
+                        next=next_link(app, request.operation, {**request.arguments, 'cursor': token}),
+                        next_requires_auth=True)
+        return HandlerOutput(data=data)
+
+    @op('communication.receipt_get', obj({'request_id': REQUEST_ID}, ('request_id',)), effect='read')
+    async def receipt_get(ctx, request, tx):
+        subject = await _subject(app, ctx, request, tx)
+        row = tx.one('SELECT body FROM results WHERE subject=? AND request_id=?',
+                     (subject, request.arguments['request_id']))
+        require(row is not None, 'not_found')
+        return HandlerOutput(data={'receipt': await _project(app, ctx, request, tx, row[0])})

src/msg/transports/http_routes.py

@@ -1384,6 +1384,59 @@ async def dispatch(request:Request):
                     require(len(payload)<=limits.max_response_bytes,'response_too_large')
                     return Response(b'' if request.method=='HEAD' else payload,
                                     media_type='application/json',headers=headers)
+                if name=='receipts':
+                    require(request.method in {'GET','HEAD'},'method_not_allowed')
+                    require(raw_path.decode('utf-8')==request.url.path and b'%' not in raw_path,
+                            'not_found')
+                    tail=(remainder or '').strip('/')
+                    listing=tail in {'','json'}
+                    if not listing:
+                        require(re.fullmatch(r'[A-Za-z0-9_.:-]{1,128}(?:/json)?',tail)
+                                is not None,'not_found')
+                    operation='communication.receipt_'+('list' if listing else 'get')
+                    pairs=request.query_params.multi_items()
+                    require(len(pairs)==len({key for key,_ in pairs}),
+                            'duplicate_query_parameter')
+                    query=dict(pairs)
+                    require(not query or listing,'unknown_query_parameter')
+                    require(set(query)<={'limit','cursor'},'unknown_query_parameter')
+                    args={} if listing else {'request_id':tail.removesuffix('/json')}
+                    if 'limit' in query:
+                        require(query['limit'].isdecimal(),'invalid_limit')
+                        args['limit']=int(query['limit'])
+                    if 'cursor' in query: args['cursor']=query['cursor']
+                    async with service.metadata.transaction(write=False) as tx:
+                        subject_id=await tx.resolve('/@'+handle)
+                        require((await tx.resource(subject_id)).type=='user','not_found')
+                    require(service.registry.operation(operation).effect=='read',
+                            'effect_mismatch')
+                    header=request.headers.get('x-msg-request')
+                    if header:
+                        packet=path_packet(header,'j',limits.max_request_bytes)
+                        require(packet.operation==operation and
+                                canonical(packet.arguments)==canonical(args),
+                                'representation_mismatch')
+                    else:
+                        packet=request_for(operation,args,service.settings.service_url,
+                                           source='manual')
+                    result=await service.executor.execute(packet,entry='network')
+                    # Another subject's lookup must not reveal whether its own
+                    # request_id exists; the path subject is the only reader.
+                    require(result.subject is None or result.subject==subject_id,
+                            'permission_denied')
+                    if result.error:
+                        return json_response(result_wire(result),error_status(result.error.code))
+                    value=wire(result.data)
+                    value['path']='/@'+handle+'/receipts'+('/'+args['request_id'] if not listing else '')
+                    etag='"'+digest(value)[7:]+'"'
+                    headers={**BASE_HEADERS,'ETag':etag,
+                             'Cache-Control':'private, no-cache'}
+                    if request.headers.get('if-none-match')==etag:
+                        return Response(status_code=304,headers=headers)
+                    payload=canonical(value)
+                    require(len(payload)<=limits.max_response_bytes,'response_too_large')
+                    return Response(b'' if request.method=='HEAD' else payload,
+                                    media_type='application/json',headers=headers)
                 certificate_detail=(name in {'cert','certificates'} and
                                     remainder not in {None,'/','/json','/meta','/history'})
                 if name in SUBJECT_KEY_ALIASES or name in SUBJECT_OPERATION_ALIASES or certificate_detail:

tests/test_subject_receipts.py

@@ -0,0 +1,193 @@
+"""/@user/receipts projects the caller's persisted OperationResults read-only."""
+from datetime import timedelta
+
+import httpx
+import pytest
+
+from msg.core.codec import b64, canonical, digest, wire
+from msg.core.requests import request_for
+from msg.transports.http import create_app
+from test_service import NOW, call, register
+
+PROJECTION = {'request_id', 'operation', 'status', 'committed_at', 'resources', 'signed'}
+
+
+def _headers(app, operation, args, key, subject):
+    packet=request_for(operation,args,app.settings.service_url,signer=key,
+                       subject=subject,expires_at=NOW+timedelta(seconds=120))
+    return {'X-Msg-Request':b64(canonical(packet))}
+
+
+def _state(tx):
+    counts=tuple(tx.one(f'SELECT COUNT(*) FROM {table}')[0] for table in
+                 ('results','audit','jobs','events','messages','revisions'))
+    results=digest([list(row) for row in tx.rows(
+        'SELECT subject,request_id,digest,body FROM results ORDER BY subject,request_id')])
+    generations=tx.one('SELECT SUM(generation) FROM resources')[0]
+    return counts,results,generations
+
+
+@pytest.mark.asyncio
+async def test_subject_receipts_are_owner_only_read_only_projection(installed):
+    app,_=installed
+    alice_key,alice,_=await register(app,'receipt-alice')
+    bob_key,bob,_=await register(app,'receipt-bob')
+    written=[]
+    for index in range(3):
+        post=await call(app,'content.post_create',{'parent':'/main','body':f'receipt {index}'},
+                        key=alice_key,subject=alice)
+        assert post.status=='ok',wire(post)
+        written.append(post)
+    bob_post=await call(app,'content.post_create',{'parent':'/main','body':'bob receipt'},
+                        key=bob_key,subject=bob)
+    assert bob_post.status=='ok',wire(bob_post)
+    async with app.metadata.transaction(write=False) as tx:
+        expected=[row[0] for row in tx.rows(
+            'SELECT request_id FROM results WHERE subject=? ORDER BY request_id',(alice,))]
+        before=_state(tx)
+    assert {post.request_id for post in written}<=set(expected) and len(expected)>=4
+    assert bob_post.request_id not in expected
+
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=create_app(app)),
+                                 base_url=app.settings.service_url) as http:
+        list_headers=_headers(app,'communication.receipt_list',{},alice_key,alice)
+        listed=await http.get('/@receipt-alice/receipts',headers=list_headers)
+        assert listed.status_code==200,listed.text
+        body=listed.json()
+        assert body['path']=='/@receipt-alice/receipts'
+        assert [item['request_id'] for item in body['items']]==expected
+        assert 'cursor' not in body
+        for item in body['items']:
+            assert set(item)==PROJECTION
+            assert item['signed'] is True and item['status']=='ok'
+        slashed=await http.get('/@receipt-alice/receipts/',headers=list_headers)
+        suffixed=await http.get('/@receipt-alice/receipts/json',headers=list_headers)
+        assert slashed.content==suffixed.content==listed.content
+        cached=await http.get('/@receipt-alice/receipts',
+                              headers={**list_headers,'If-None-Match':listed.headers['etag']})
+        assert cached.status_code==304
+
+        target=written[1]
+        item_headers=_headers(app,'communication.receipt_get',
+                              {'request_id':target.request_id},alice_key,alice)
+        single=await http.get(f'/@receipt-alice/receipts/{target.request_id}',headers=item_headers)
+        assert single.status_code==200,single.text
+        receipt=single.json()['receipt']
+        assert single.json()['path']==f'/@receipt-alice/receipts/{target.request_id}'
+        assert receipt==next(item for item in body['items']
+                             if item['request_id']==target.request_id)
+        assert receipt['operation']=='content.post_create'
+        assert receipt['resources']==[wire(ref) for ref in target.resources]
+        assert receipt['committed_at']==wire(target.committed_at)
+        for secret in ('data','output','receipt','cli_url','prefer_cli','actor','subject'):
+            assert secret not in receipt
+        suffixed=await http.get(f'/@receipt-alice/receipts/{target.request_id}/json',
+                                headers=item_headers)
+        assert suffixed.content==single.content
+        head=await http.head(f'/@receipt-alice/receipts/{target.request_id}',headers=item_headers)
+        assert head.status_code==200 and head.content==b''
+        assert head.headers['etag']==single.headers['etag']
+
+        missing_headers=_headers(app,'communication.receipt_get',
+                                 {'request_id':'r_missing_receipt'},alice_key,alice)
+        missing=await http.get('/@receipt-alice/receipts/r_missing_receipt',headers=missing_headers)
+        assert missing.status_code==404 and missing.json()['error']['code']=='not_found'
+
+        bob_list=_headers(app,'communication.receipt_list',{},bob_key,bob)
+        cross=await http.get('/@receipt-alice/receipts',headers=bob_list)
+        assert cross.status_code==403 and cross.json()['error']['code']=='permission_denied'
+        assert target.request_id not in cross.text
+        for request_id in (target.request_id,bob_post.request_id,'r_missing_receipt'):
+            bob_get=_headers(app,'communication.receipt_get',{'request_id':request_id},bob_key,bob)
+            denied=await http.get(f'/@receipt-alice/receipts/{request_id}',headers=bob_get)
+            assert denied.status_code==403,denied.text
+            assert denied.json()['error']['code']=='permission_denied'
+        own=await http.get('/@receipt-bob/receipts',headers=bob_list)
+        assert own.status_code==200
+        assert bob_post.request_id in [item['request_id'] for item in own.json()['items']]
+        assert not set(expected)&{item['request_id'] for item in own.json()['items']}
+
+        for path in ('/@receipt-alice/receipts',f'/@receipt-alice/receipts/{target.request_id}'):
+            anonymous=await http.get(path)
+            assert anonymous.status_code==401
+            assert anonymous.json()['error']['code']=='authentication_required'
+            assert 'items' not in anonymous.json() and 'receipt' not in anonymous.json()
+
+        unknown=await http.get('/@receipt-alice/receipts?after=x',headers=list_headers)
+        assert unknown.status_code==400
+        assert unknown.json()['error']['code']=='unknown_query_parameter'
+        detail_query=await http.get(f'/@receipt-alice/receipts/{target.request_id}?limit=1',
+                                    headers=item_headers)
+        assert detail_query.json()['error']['code']=='unknown_query_parameter'
+        for bad in (0,201):
+            bad_headers=_headers(app,'communication.receipt_list',{'limit':bad},alice_key,alice)
+            rejected=await http.get(f'/@receipt-alice/receipts?limit={bad}',headers=bad_headers)
+            assert rejected.status_code==400,rejected.text
+        for path in ('/@receipt-alice/receipts',f'/@receipt-alice/receipts/{target.request_id}'):
+            written_path=await http.post(path,json={})
+            assert written_path.status_code in {404,405}
+
+        pages=[]
+        cursor=None
+        while True:
+            args={'limit':1} if cursor is None else {'limit':1,'cursor':cursor}
+            page=await http.get('/@receipt-alice/receipts',params=args,
+                                headers=_headers(app,'communication.receipt_list',args,
+                                                 alice_key,alice))
+            assert page.status_code==200,page.text
+            data=page.json()
+            assert len(data['items'])==1
+            pages.append(data['items'][0]['request_id'])
+            cursor=data.get('cursor')
+            if cursor is None:
+                break
+            assert data['next_requires_auth'] is True
+            assert data['next'].startswith('/-/g/communication.receipt_list/j/')
+            stolen_args={'limit':1,'cursor':cursor}
+            stolen=await http.get('/@receipt-bob/receipts',params=stolen_args,
+                                  headers=_headers(app,'communication.receipt_list',
+                                                   stolen_args,bob_key,bob))
+            assert stolen.status_code==400
+            assert stolen.json()['error']['code']=='cursor_query_mismatch'
+        assert pages==expected
+
+    async with app.metadata.transaction(write=False) as tx:
+        assert _state(tx)==before
+
+
+@pytest.mark.asyncio
+async def test_receipt_projection_hides_resources_no_longer_readable(installed):
+    app,_=installed
+    alice_key,alice,_=await register(app,'receipt-share-alice')
+    bob_key,bob,_=await register(app,'receipt-share-bob')
+    note=await call(app,'identity.note_put',{'name':'receipt-note','body':'Private'},
+                    key=alice_key,subject=alice)
+    assert note.status=='ok',wire(note)
+    rid=note.resources[0].id
+    grant=await call(app,'sharing.grant',{'resource':rid,'grantee':bob,
+                                          'expires_at':wire(NOW+timedelta(days=1))},
+                     key=alice_key,subject=alice)
+    assert grant.status=='ok',wire(grant)
+    sent=await call(app,'communication.send',
+                       {'recipient':alice,'resource':wire(note.resources[0])},
+                       key=bob_key,subject=bob)
+    assert sent.status=='ok',wire(sent)
+    assert rid in {ref.id for ref in sent.resources}
+    shown=await call(app,'communication.receipt_get',{'request_id':sent.request_id},
+                     key=bob_key,subject=bob)
+    assert shown.status=='ok',wire(shown)
+    assert rid in {ref['id'] for ref in shown.data['receipt']['resources']}
+    revoked=await call(app,'sharing.revoke',{'grant_id':grant.data['grant']['id']},
+                       key=alice_key,subject=alice)
+    assert revoked.status=='ok',wire(revoked)
+    async with app.metadata.transaction(write=False) as tx:
+        stored=tx.one('SELECT body FROM results WHERE subject=? AND request_id=?',
+                      (bob,sent.request_id))[0]
+    hidden=await call(app,'communication.receipt_get',{'request_id':sent.request_id},
+                      key=bob_key,subject=bob)
+    assert hidden.status=='ok',wire(hidden)
+    assert rid not in {ref['id'] for ref in hidden.data['receipt']['resources']}
+    assert hidden.data['receipt']['signed'] is True
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.one('SELECT body FROM results WHERE subject=? AND request_id=?',
+                      (bob,sent.request_id))[0]==stored