Fix scoped read, content, authorization and delivery contract defects

src/msg/core/registry.py

@@ -37,7 +37,10 @@ def __init__(self):
 
     def _insert(self, collection, spec, kind):
         require(not self._frozen, 'registry_frozen')
-        require(spec.version >= 1 and spec.name and '*' not in spec.name, 'invalid_registry_name')
+        require(
+            type(spec.version) is int and spec.version >= 1 and spec.name and '*' not in spec.name,
+            'invalid_registry_name',
+        )
         key = (spec.name, spec.version)
         require(key not in collection, 'duplicate_' + kind)
         collection[key] = spec
@@ -126,7 +129,11 @@ def add(self, manifest):
             )
         for item in (*manifest.resource_types, *manifest.capabilities, *manifest.operations):
             require(
-                item.version >= 1 and item.name and '*' not in item.name, 'invalid_registry_name'
+                type(item.version) is int
+                and item.version >= 1
+                and item.name
+                and '*' not in item.name,
+                'invalid_registry_name',
             )
         for item in manifest.operations:
             require(

src/msg/core/text_patch.py

@@ -150,6 +150,9 @@ def _hunks(text):
                 previous = None
             else:
                 require(line[:1] in {' ', '-', '+'}, 'invalid_unified_patch')
+                # A missing file newline must be represented by the explicit
+                # marker, not an unterminated diff transport line.
+                require(line.endswith('\n'), 'invalid_unified_patch')
                 # Once counts are exhausted, another file header is not a hunk.
                 require(len(old) < old_count or len(new) < new_count, 'invalid_unified_patch')
                 previous = []
@@ -195,6 +198,12 @@ def _unified(source, hunks, *, rebase=False):
         require(last <= start <= len(lines), 'invalid_unified_patch')
         end = start + len(hunk.old)
         require(tuple(lines[start:end]) == hunk.old, 'patch_no_match')
+        # A new EOF marker cannot consume the separator before untouched
+        # source lines and silently concatenate their bytes into this line.
+        require(
+            not hunk.new or hunk.new[-1].endswith('\n') or end == len(lines),
+            'invalid_unified_patch',
+        )
         result.extend(lines[last:start])
         result.extend(hunk.new)
         last = end

src/msg/extensions/hosting.py

@@ -40,11 +40,17 @@ def path_name(value):
         and all(ord(c) >= 32 and ord(c) != 127 for c in value),
         'invalid_hosting_path',
     )
+    # Hosted reads deliberately reject percent-encoded and non-ASCII paths.
+    # Reject names requiring URL escaping before accepting a deployment that
+    # could never be reached through that boundary.
+    require(quote(value, safe="/:@!$&'()*+,;=") == value, 'invalid_hosting_path')
     parts = value.split('/')
     require(
         all(part not in {'', '.', '..'} and not part.startswith('.') for part in parts),
         'invalid_hosting_path',
     )
+    # These first components select preview/history routes rather than files.
+    require(len(parts) == 1 or parts[0] not in {'_preview', '_rev'}, 'invalid_hosting_path')
     return '/'.join(parts)
 
 

src/msg/plugins/bounty.py

@@ -135,6 +135,16 @@ async def validate_contract(app, tx, row):
     """Mutable accounting state cannot change a publisher's immutable terms."""
     from msg.market.catalog import read_listing_body
 
+    # A mixed restore must not redirect payouts, refunds or top-ups to another
+    # listing's escrow or a signable Subject account.
+    account = tx.one(
+        'SELECT kind,subject_id,source_id FROM ledger_accounts WHERE id=?',
+        (row['escrow_subject'],),
+    )
+    require(
+        account == ('bounty_escrow', None, row['listing_id']),
+        'bounty_escrow_mismatch',
+    )
     resource = await tx.resource(row['listing_id'])
     body, _ = await read_listing_body(app, tx, resource)
     require(

src/msg/plugins/discovery.py

@@ -1673,7 +1673,10 @@ async def endpoint_visible(rid):
                 else any(term in whole for term in terms)
             ):
                 continue
-            if exact and exact not in whole:
+            # Fields are separate documents for phrase matching. Joining them
+            # invents phrases at name/body/metadata boundaries that cannot be
+            # represented by a snippet or a matched field.
+            if exact and not any(exact in value for value in lowered.values()):
                 continue
             if any(term in whole for term in excluded):
                 continue

src/msg/security/authorization.py

@@ -196,17 +196,20 @@ async def shared_read(self, principal, resource, chain, now, session):
         if share_target_error(self.registry, resource, chain, session):
             return False
         row = session.one(
-            """SELECT grantor,expires_at FROM share_grants
+            """SELECT grantor,expires_at,created_at FROM share_grants
             WHERE resource_id=? AND grantee=? AND revoked_at IS NULL""",
             (resource.id, principal.subject),
         )
-        if (
-            row
-            and row[0] == resource.owner
-            and parse_time(row[1]) > now
-            and resource.state == 'active'
-        ):
-            return True
+        if row and row[0] == resource.owner and resource.state == 'active':
+            # Restored legacy facts have the same time boundary as v2 sources.
+            # A malformed source denies only itself, preserving independent ones.
+            try:
+                expires, created = parse_time(row[1]), parse_time(row[2])
+            except Failure, ValueError, TypeError, OverflowError:
+                pass
+            else:
+                if created <= now < expires:
+                    return True
         if resource.state != 'active':
             return False
         candidates = session.rows(

src/msg/security/network.py

@@ -59,7 +59,9 @@ def validate_url(url, method, policy):
             'invalid_network_url',
         )
         host = normalized_host(parsed.hostname)
-        port = parsed.port or (443 if parsed.scheme == 'https' else 80)
+        port = parsed.port
+        if port is None:
+            port = 443 if parsed.scheme == 'https' else 80
     except ValueError as exc:
         if isinstance(exc, Failure):
             raise

src/msg/transports/http_routes.py

@@ -821,7 +821,7 @@ async def dispatch(request: Request):
                 args = dict(request.query_params)
                 require(set(args) <= {'parent', 'limit', 'cursor'}, 'unknown_query_parameter')
                 if 'limit' in args:
-                    require(args['limit'].isdigit(), 'invalid_limit')
+                    require(args['limit'].isdecimal(), 'invalid_limit')
                     args['limit'] = int(args['limit'])
                 header = request.headers.get('x-msg-request')
                 packet = (

src/msg/workers/webhook.py

@@ -56,7 +56,7 @@ def validate_endpoint(url: str) -> tuple[str, int]:
             'invalid_webhook_url',
         )
         host = normalized_host(parsed.hostname)
-        port = parsed.port or 443
+        port = 443 if parsed.port is None else parsed.port
     except ValueError as exc:
         raise Failure('invalid_webhook_url') from exc
     require(port == 443, 'webhook_port_forbidden')

tests/test_achievement_concurrency.py

@@ -0,0 +1,125 @@
+"""Concurrent and retried ceremonies keep one challenge and one honor fact."""
+
+import asyncio
+
+import pytest
+from test_achievements import advance
+from test_service import call, register
+
+from msg.core.codec import loads, wire
+from msg.plugins.achievements import FINAL_STATEMENT
+
+
+@pytest.mark.asyncio
+async def test_concurrent_starts_keep_the_successful_challenge_usable(installed):
+    app, _ = installed
+    key, subject, _ = await register(app, 'honor-concurrent-start')
+    results = await asyncio.gather(
+        *(
+            call(app, 'achievement.start', {}, key=key, subject=subject, rid=f'start-{index}')
+            for index in range(2)
+        )
+    )
+    successes = [result for result in results if result.status == 'ok']
+    errors = [result for result in results if result.status == 'error']
+    assert len(successes) == len(errors) == 1, [wire(result) for result in results]
+    assert errors[0].error.code == 'achievement_ceremony_active'
+    async with app.metadata.transaction(write=False) as tx:
+        rows = tx.rows('SELECT id,state FROM achievement_ceremonies WHERE subject=?', (subject,))
+        assert rows == [(successes[0].data['challenge_id'], 'active')]
+    advanced = await advance(app, key, subject, successes[0].data)
+    assert advanced.status == 'ok' and advanced.data['round'] == 2, wire(advanced)
+
+
+@pytest.mark.asyncio
+async def test_answer_request_replay_preserves_nonce_without_advancing_again(installed):
+    app, _ = installed
+    key, subject, _ = await register(app, 'honor-answer-replay')
+    challenge = (await call(app, 'achievement.start', {}, key=key, subject=subject)).data
+    arguments = {
+        name: challenge[name] for name in ('challenge_id', 'round', 'question_digest', 'nonce')
+    }
+    arguments['answer'] = 'y'
+    results = await asyncio.gather(
+        *(
+            call(app, 'achievement.answer', arguments, key=key, subject=subject, rid='answer-once')
+            for _ in range(2)
+        )
+    )
+    assert all(result.status == 'ok' for result in results), [wire(result) for result in results]
+    assert sum(result.replayed for result in results) == 1
+    assert results[0].data == results[1].data
+    assert results[0].data['round'] == 2
+    assert results[0].data['nonce'] != challenge['nonce']
+    third = await advance(app, key, subject, results[0].data)
+    assert third.status == 'ok' and third.data['round'] == 3, wire(third)
+    replay = await call(
+        app, 'achievement.answer', arguments, key=key, subject=subject, rid='answer-once'
+    )
+    assert replay.status == 'ok' and replay.replayed
+    assert replay.data == results[0].data
+    async with app.metadata.transaction(write=False) as tx:
+        state = loads(
+            tx.one(
+                'SELECT body FROM achievement_ceremonies WHERE id=?', (challenge['challenge_id'],)
+            )[0]
+        )
+        assert state['status'] == 'active' and state['round'] == 3
+        assert [answer['round'] for answer in state['answers']] == [1, 2]
+
+
+@pytest.mark.asyncio
+async def test_concurrent_finish_and_replay_issue_only_one_grant_and_audit(installed):
+    app, _ = installed
+    key, subject, _ = await register(app, 'honor-concurrent-finish')
+    challenge = (await call(app, 'achievement.start', {}, key=key, subject=subject)).data
+    for _ in range(4):
+        result = await advance(app, key, subject, challenge)
+        assert result.status == 'ok', wire(result)
+        challenge = result.data
+    arguments = {
+        name: challenge[name]
+        for name in ('challenge_id', 'question_digest', 'nonce', 'ceremony_digest')
+    }
+    arguments['statement'] = FINAL_STATEMENT
+    results = await asyncio.gather(
+        *(
+            call(
+                app,
+                'achievement.finish',
+                arguments,
+                key=key,
+                subject=subject,
+                rid=f'finish-{index}',
+            )
+            for index in range(2)
+        )
+    )
+    successful = [index for index, result in enumerate(results) if result.status == 'ok']
+    assert len(successful) == 1, [wire(result) for result in results]
+    winner = successful[0]
+    assert results[1 - winner].error.code == 'ceremony_inactive'
+    replay = await call(
+        app,
+        'achievement.finish',
+        arguments,
+        key=key,
+        subject=subject,
+        rid=f'finish-{winner}',
+    )
+    assert replay.status == 'ok' and replay.replayed
+    assert replay.data == results[winner].data
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.one('SELECT COUNT(*) FROM achievement_grants WHERE subject=?', (subject,))[0] == 1
+        audits = [loads(row[0]) for row in tx.rows('SELECT body FROM audit')]
+    issued = [
+        audit
+        for audit in audits
+        if audit['event']['type'] == 'achievement.auto.issue'
+        and audit['event']['subject'] == subject
+    ]
+    assert len(issued) == 1
+    assert issued[0]['event']['data']['certificate_id'] == replay.data['grant']['id']
+    listed = await call(app, 'achievement.list', {'subject_id': subject})
+    assert listed.status == 'ok' and len(listed.data['achievements']) == 1
+    assert listed.data['achievements'][0] == replay.data['grant']

tests/test_bounty_escrow_binding.py

@@ -0,0 +1,68 @@
+"""Mixed restored projections cannot spend an unrelated ledger account."""
+
+import pytest
+from test_bounty import fund, terms
+from test_market_72 import proof
+from test_service import call, register
+
+from msg.core.codec import wire
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('target', ['subject', 'other_bounty'])
+@pytest.mark.parametrize(
+    'operation', ['bounty.claim', 'bounty.close', 'bounty.top_up', 'bounty.get']
+)
+async def test_bounty_escrow_must_belong_to_its_listing(installed, target, operation):
+    app, _ = installed
+    publisher_key, publisher, _ = await register(app, 'escrow-binding-publisher')
+    claimant_key, claimant, _ = await register(app, 'escrow-binding-claimant')
+    await fund(app, publisher, 40)
+    created = await call(app, 'bounty.create', terms(), key=publisher_key, subject=publisher)
+    assert created.status == 'ok', wire(created)
+    listing_id = created.data['bounty']['listing_id']
+    claim = await proof(app, (claimant_key, claimant), listing_id)
+    wrong_account = publisher
+    if target == 'other_bounty':
+        other = await call(
+            app, 'bounty.create', terms(name='other-escrow'), key=publisher_key, subject=publisher
+        )
+        assert other.status == 'ok', wire(other)
+        wrong_account = other.data['funding']['body']['to_subject']
+    async with app.metadata.transaction(write=True) as tx:
+        if target == 'other_bounty':
+            # Keep the unique escrow pointer constraint intact while emulating
+            # inconsistent materialized rows from different restored snapshots.
+            tx.execute(
+                'UPDATE bounty_listings SET escrow_subject=? WHERE listing_id=?',
+                (publisher, other.data['bounty']['listing_id']),
+                write=True,
+            )
+        tx.execute(
+            'UPDATE bounty_listings SET escrow_subject=? WHERE listing_id=?',
+            (wrong_account, listing_id),
+            write=True,
+        )
+        before_ledger = tx.rows('SELECT * FROM money_ledger ORDER BY id')
+        before_challenge = tx.one(
+            'SELECT * FROM bounty_challenges WHERE id=?', (claim['challenge_id'],)
+        )
+    if operation == 'bounty.claim':
+        params, key, subject = claim, claimant_key, claimant
+    else:
+        params, key, subject = {'listing_id': listing_id}, publisher_key, publisher
+        if operation == 'bounty.top_up':
+            params['amount_minor'] = 1
+    result = await call(app, operation, params, key=key, subject=subject)
+    assert result.status == 'error' and result.error.code == 'bounty_escrow_mismatch', wire(result)
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.rows('SELECT * FROM money_ledger ORDER BY id') == before_ledger
+        assert (
+            tx.one('SELECT * FROM bounty_challenges WHERE id=?', (claim['challenge_id'],))
+            == before_challenge
+        )
+        assert tx.one('SELECT COUNT(*) FROM bounty_claims')[0] == 0
+        assert (
+            tx.one("SELECT COUNT(*) FROM messages WHERE body::jsonb->>'source'='bounty_claim'")[0]
+            == 0
+        )