diff --git a/src/msg/core/registry.py b/src/msg/core/registry.py
index 2fbaeb62..ce00a4c6 100644
--- a/src/msg/core/registry.py
+++ b/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(
diff --git a/src/msg/core/text_patch.py b/src/msg/core/text_patch.py
index 7dc31c43..9504333e 100644
--- a/src/msg/core/text_patch.py
+++ b/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
diff --git a/src/msg/extensions/hosting.py b/src/msg/extensions/hosting.py
index 98db6719..ec5aee16 100644
--- a/src/msg/extensions/hosting.py
+++ b/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)
diff --git a/src/msg/plugins/bounty.py b/src/msg/plugins/bounty.py
index 9b511ce2..25111a9f 100644
--- a/src/msg/plugins/bounty.py
+++ b/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(
diff --git a/src/msg/plugins/discovery.py b/src/msg/plugins/discovery.py
index ad5f4258..4148c278 100644
--- a/src/msg/plugins/discovery.py
+++ b/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
diff --git a/src/msg/security/authorization.py b/src/msg/security/authorization.py
index 9e27e30b..4a22c0d4 100644
--- a/src/msg/security/authorization.py
+++ b/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(
diff --git a/src/msg/security/network.py b/src/msg/security/network.py
index f10d4c0b..03f43f84 100644
--- a/src/msg/security/network.py
+++ b/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
diff --git a/src/msg/transports/http_routes.py b/src/msg/transports/http_routes.py
index b253d69e..9d259374 100644
--- a/src/msg/transports/http_routes.py
+++ b/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 = (
diff --git a/src/msg/workers/webhook.py b/src/msg/workers/webhook.py
index 2a331a1a..f63fbb98 100644
--- a/src/msg/workers/webhook.py
+++ b/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')
diff --git a/tests/test_achievement_concurrency.py b/tests/test_achievement_concurrency.py
new file mode 100644
index 00000000..f5992435
--- /dev/null
+++ b/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']
diff --git a/tests/test_bounty_escrow_binding.py b/tests/test_bounty_escrow_binding.py
new file mode 100644
index 00000000..14da84a7
--- /dev/null
+++ b/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
+ )
diff --git a/tests/test_hosting_addressable_paths.py b/tests/test_hosting_addressable_paths.py
new file mode 100644
index 00000000..183528d1
--- /dev/null
+++ b/tests/test_hosting_addressable_paths.py
@@ -0,0 +1,107 @@
+"""Deployments reject names the hosted HTTP boundary cannot address."""
+
+import httpx
+import pytest
+from test_service import call, register
+
+from msg.core.codec import b64, wire
+from msg.transports.http import create_app
+
+
+@pytest.mark.parametrize('operation', ['hosting.deploy', 'hosting.preview'])
+@pytest.mark.asyncio
+async def test_unaddressable_paths_are_rejected_before_materialization(installed, operation):
+ app, _ = installed
+ key, user, _ = await register(app, 'path-owner')
+ site = await call(
+ app, 'hosting.create', {'parent': '/@path-owner', 'name': 'web'}, key=key, subject=user
+ )
+ source = await call(
+ app,
+ 'content.file_put',
+ {
+ 'parent': '/@path-owner/files',
+ 'name': 'page.html',
+ 'data': b64(b'
addressable
'),
+ 'media_type': 'text/html',
+ },
+ key=key,
+ subject=user,
+ )
+ assert site.status == source.status == 'ok'
+ async with app.metadata.transaction(write=False) as tx:
+ before = tuple(
+ tx.one(f'SELECT COUNT(*) FROM {table}')[0]
+ for table in ('resources', 'revisions', 'settings')
+ )
+ for path in (
+ 'hello world.html',
+ '中文.html',
+ 'a%20b.html',
+ 'a#b.html',
+ 'a?b.html',
+ '_preview/fake/index.html',
+ '_rev/fake/index.html',
+ ):
+ result = await call(
+ app,
+ operation,
+ {
+ 'id': site.resources[0].id,
+ 'entries': [{'path': path, 'source': wire(source.resources[0])}],
+ },
+ key=key,
+ subject=user,
+ expected=((site.resources[0].id, site.data['generation']),),
+ )
+ assert result.status == 'error' and result.error.code == 'invalid_hosting_path', wire(
+ result
+ )
+ async with app.metadata.transaction(write=False) as tx:
+ assert (
+ tuple(
+ tx.one(f'SELECT COUNT(*) FROM {table}')[0]
+ for table in ('resources', 'revisions', 'settings')
+ )
+ == before
+ )
+
+
+@pytest.mark.parametrize('path', ["assets/v1/page-_.~!$&'()*+,;=:@.html", '_rev', '_preview'])
+@pytest.mark.asyncio
+async def test_uri_safe_paths_remain_reachable(installed, path):
+ app, _ = installed
+ key, user, _ = await register(app, 'safe-path-owner')
+ site = await call(
+ app, 'hosting.create', {'parent': '/@safe-path-owner', 'name': 'web'}, key=key, subject=user
+ )
+ html = b'nested path
'
+ source = await call(
+ app,
+ 'content.file_put',
+ {
+ 'parent': '/@safe-path-owner/files',
+ 'name': 'page.html',
+ 'data': b64(html),
+ 'media_type': 'text/html',
+ },
+ key=key,
+ subject=user,
+ )
+ deployed = await call(
+ app,
+ 'hosting.deploy',
+ {
+ 'id': site.resources[0].id,
+ 'entries': [{'path': path, 'source': wire(source.resources[0])}],
+ },
+ key=key,
+ subject=user,
+ expected=((site.resources[0].id, site.data['generation']),),
+ )
+ assert deployed.status == 'ok', wire(deployed)
+ async with httpx.AsyncClient(
+ transport=httpx.ASGITransport(app=create_app(app)), base_url='http://testserver'
+ ) as http:
+ served = await http.get('/@safe-path-owner/web/' + path)
+ assert served.status_code == 200 and served.content == html
diff --git a/tests/test_legacy_share_source_isolation.py b/tests/test_legacy_share_source_isolation.py
new file mode 100644
index 00000000..26734197
--- /dev/null
+++ b/tests/test_legacy_share_source_isolation.py
@@ -0,0 +1,53 @@
+"""Restored legacy shares fail closed without vetoing an independent source."""
+
+from datetime import timedelta
+
+import pytest
+from test_authorization_sources import grant, private_post
+from test_service import NOW, register
+from test_share_grants_v2 import invoke, ok
+
+from msg.core.codec import wire
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+ ('column', 'value'),
+ [
+ ('expires_at', 'corrupt-restored-time'),
+ ('created_at', 'corrupt-restored-time'),
+ ('created_at', wire(NOW + timedelta(hours=1))),
+ ],
+)
+async def test_malformed_legacy_source_does_not_veto_independent_share(installed, column, value):
+ app, _ = installed
+ owner = await register(app, 'legacy-source-owner')
+ reader = await register(app, 'legacy-source-reader')
+ rid, revision = await private_post(app, owner)
+ legacy = ok(
+ await invoke(
+ app,
+ owner,
+ 'sharing.grant',
+ {'resource': rid, 'grantee': reader[1], 'expires_at': wire(NOW + timedelta(days=1))},
+ )
+ ).data['grant']['id']
+ queries = ({'id': rid}, {'id': rid, 'view': 'history'}, {'id': rid, 'revision': revision})
+ for query in queries:
+ ok(await invoke(app, reader, 'discovery.get', query))
+
+ async with app.metadata.transaction(write=True) as tx:
+ tx.execute(f'UPDATE share_grants SET {column}=? WHERE id=?', (value, legacy), write=True)
+ for query in queries:
+ denied = await invoke(app, reader, 'discovery.get', query)
+ assert denied.status == 'error' and denied.error.code == 'permission_denied', wire(denied)
+ assert not denied.resources and not denied.data
+
+ independent = await grant(app, owner, rid, reader[1])
+ for query in queries:
+ ok(await invoke(app, reader, 'discovery.get', query))
+ ok(await invoke(app, owner, 'sharing.revoke', {'grant_id': independent}, version=2))
+ for query in queries:
+ denied = await invoke(app, reader, 'discovery.get', query)
+ assert denied.status == 'error' and denied.error.code == 'permission_denied', wire(denied)
+ assert not denied.resources and not denied.data
diff --git a/tests/test_network_url_ports.py b/tests/test_network_url_ports.py
new file mode 100644
index 00000000..b6b57b55
--- /dev/null
+++ b/tests/test_network_url_ports.py
@@ -0,0 +1,45 @@
+from dataclasses import replace
+
+import pytest
+
+from msg.core.errors import Failure
+from msg.core.models import NetworkPolicy
+from msg.security.network import validate_url
+
+
+@pytest.fixture
+def policy():
+ return NetworkPolicy(
+ schemes=frozenset({'http', 'https'}),
+ hosts=(),
+ ports=frozenset({80, 443}),
+ methods=frozenset({'GET'}),
+ allow_private=False,
+ timeout_ms=1000,
+ max_response_bytes=10000,
+ max_redirects=3,
+ )
+
+
+@pytest.mark.parametrize('scheme', ['http', 'https'])
+def test_explicit_zero_port_does_not_inherit_scheme_default(scheme, policy):
+ with pytest.raises(Failure, match='network_port_forbidden'):
+ validate_url(f'{scheme}://example.org:0/path', 'GET', policy)
+
+
+@pytest.mark.parametrize('port', ['-1', '65536', 'invalid'])
+def test_invalid_explicit_port_is_a_network_url_failure(port, policy):
+ with pytest.raises(Failure, match='invalid_network_url'):
+ validate_url(f'https://example.org:{port}/path', 'GET', policy)
+
+
+@pytest.mark.parametrize(('scheme', 'port'), [('http', 80), ('https', 443)])
+def test_default_port_and_explicit_port_obey_same_allowlist(scheme, port, policy):
+ implicit = f'{scheme}://example.org/path'
+ explicit = f'{scheme}://example.org:{port}/path'
+ assert validate_url(implicit, 'GET', policy).port is None
+ assert validate_url(explicit, 'GET', policy).port == port
+ denied = replace(policy, ports=frozenset({8443}))
+ for url in (implicit, explicit):
+ with pytest.raises(Failure, match='network_port_forbidden'):
+ validate_url(url, 'GET', denied)
diff --git a/tests/test_registry_contract_references.py b/tests/test_registry_contract_references.py
index 55819849..b1ca40f9 100644
--- a/tests/test_registry_contract_references.py
+++ b/tests/test_registry_contract_references.py
@@ -1,6 +1,7 @@
"""Registry freeze rejects dangling references before serving any contract."""
-from dataclasses import replace
+from dataclasses import fields, replace
+from types import SimpleNamespace
import pytest
@@ -54,6 +55,14 @@ def declarations():
return resource, capability, operation
+def contract_adapter(spec, version):
+ # Normal records already reject malformed versions in codec.record.
+ # Registry also accepts plugin-provided objects, so test that boundary
+ # without asking the record constructor to accept an invalid contract.
+ values = {field.name: getattr(spec, field.name) for field in fields(spec)}
+ return SimpleNamespace(**(values | {'version': version}))
+
+
def registry(resource, capability, operation):
result = Registry()
result.add_schema(REF, {'type': 'object'})
@@ -108,3 +117,48 @@ def test_freeze_accepts_registered_content_constraints_and_operation_references(
assert installed.resource_type('document') == resource
assert installed.capability('document.read') == capability
assert installed.operation('document.read') == operation
+
+
+@pytest.mark.parametrize('kind', ['resource_type', 'capability', 'operation'])
+@pytest.mark.parametrize('version', [True, False, 1.0, 1.5, '1', None, 0, -1])
+def test_registration_rejects_non_positive_integer_versions_without_inserting(kind, version):
+ resource, capability, operation = declarations()
+ spec = {'resource_type': resource, 'capability': capability, 'operation': operation}[kind]
+ installed = Registry()
+ register = getattr(installed, 'add_' + kind)
+ with pytest.raises(Failure, match='^invalid_registry_name$'):
+ register(contract_adapter(spec, version))
+ register(spec)
+ assert getattr(installed, kind)(spec.name) == spec
+
+
+@pytest.mark.parametrize('kind', ['resource_type', 'capability', 'operation'])
+@pytest.mark.parametrize('version', [True, 1.0, '1', None])
+def test_manifest_rejects_invalid_versions_before_registering_any_contract(kind, version):
+ resource, capability, operation = declarations()
+ values = {'resource_type': resource, 'capability': capability, 'operation': operation}
+ values[kind] = contract_adapter(values[kind], version)
+ installed = Registry()
+ manifest = PluginManifest(
+ name='identity',
+ version='1',
+ dependencies=(),
+ resource_types=(values['resource_type'],),
+ capabilities=(values['capability'],),
+ operations=(values['operation'],),
+ migrations=(),
+ )
+ with pytest.raises(Failure, match='^invalid_registry_name$'):
+ installed.add(manifest)
+ assert installed.resource_types() == ()
+ assert installed.capabilities() == ()
+ assert installed.operations() == ()
+ installed.add(
+ replace(
+ manifest,
+ resource_types=(resource,),
+ capabilities=(capability,),
+ operations=(operation,),
+ )
+ )
+ assert installed.operation(operation.name) == operation
diff --git a/tests/test_rss_query_validation.py b/tests/test_rss_query_validation.py
new file mode 100644
index 00000000..8dbb4af7
--- /dev/null
+++ b/tests/test_rss_query_validation.py
@@ -0,0 +1,43 @@
+"""RSS reader errors remain client errors on the real HTTP adapter."""
+
+import httpx
+import pytest
+
+from msg.transports.http import create_app
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('path', ['/rss', '/-/rss'])
+async def test_rss_rejects_nondecimal_digits_without_internal_error(installed, path):
+ app, _ = installed
+ async with app.metadata.transaction(write=False) as tx:
+ before = tx.one('SELECT COUNT(*) FROM events')[0]
+ async with httpx.AsyncClient(
+ transport=httpx.ASGITransport(app=create_app(app)), base_url=app.settings.service_url
+ ) as http:
+ for limit in ('²', '³', '①', '1²'):
+ response = await http.get(path, params={'limit': limit})
+ assert response.status_code == 400
+ assert response.json()['error']['code'] == 'invalid_limit'
+ async with app.metadata.transaction(write=False) as tx:
+ assert tx.one('SELECT COUNT(*) FROM events')[0] == before
+
+
+@pytest.mark.asyncio
+async def test_rss_keeps_decimal_limits_and_head_read_only(installed):
+ app, _ = installed
+ async with app.metadata.transaction(write=False) as tx:
+ before = tx.one('SELECT COUNT(*) FROM events')[0]
+ async with httpx.AsyncClient(
+ transport=httpx.ASGITransport(app=create_app(app)), base_url=app.settings.service_url
+ ) as http:
+ for limit in ('1', '01', '١'):
+ response = await http.get('/rss', params={'limit': limit})
+ assert response.status_code == 200
+ assert response.headers['content-type'].startswith('application/rss+xml')
+ head = await http.head('/rss', params={'limit': limit})
+ assert head.status_code == 200
+ assert not head.content
+ assert head.headers['content-length'] == response.headers['content-length']
+ async with app.metadata.transaction(write=False) as tx:
+ assert tx.one('SELECT COUNT(*) FROM events')[0] == before
diff --git a/tests/test_search_phrase_boundaries.py b/tests/test_search_phrase_boundaries.py
new file mode 100644
index 00000000..198978bf
--- /dev/null
+++ b/tests/test_search_phrase_boundaries.py
@@ -0,0 +1,54 @@
+"""Exact phrases must occur within a real searchable field."""
+
+import pytest
+from test_service import call, register
+
+from msg.core.codec import wire
+
+
+@pytest.mark.asyncio
+async def test_exact_phrase_does_not_cross_name_body_boundary(installed):
+ app, _ = installed
+ key, subject, _ = await register(app, 'phrase-boundary-owner')
+ split = await call(
+ app,
+ 'content.post_create',
+ {'parent': '/main', 'name': 'boundaryalpha', 'body': 'boundarybeta'},
+ key=key,
+ subject=subject,
+ )
+ complete = await call(
+ app,
+ 'content.post_create',
+ {'parent': '/main', 'name': 'complete', 'body': 'boundaryalpha.md boundarybeta'},
+ key=key,
+ subject=subject,
+ )
+ assert split.status == complete.status == 'ok'
+ result = await call(
+ app,
+ 'discovery.lexical_search',
+ {
+ 'scope': '/main',
+ 'exact': 'boundaryalpha.md boundarybeta',
+ 'snippet': True,
+ 'explain': 'compact',
+ },
+ )
+ assert result.status == 'ok', wire(result)
+ assert [item['id'] for item in result.data['items']] == [complete.resources[0].id]
+ item = result.data['items'][0]
+ assert item['snippet']['field'] == 'body'
+ assert list(item['rank_reason']['matched_fields']) == ['body']
+
+ # Independent terms can still match different fields of the same resource.
+ terms = await call(
+ app,
+ 'discovery.lexical_search',
+ {'scope': '/main', 'terms': 'boundaryalpha boundarybeta', 'mode': 'all'},
+ )
+ assert terms.status == 'ok', wire(terms)
+ assert {item['id'] for item in terms.data['items']} == {
+ split.resources[0].id,
+ complete.resources[0].id,
+ }
diff --git a/tests/test_text_patch_eof_boundaries.py b/tests/test_text_patch_eof_boundaries.py
new file mode 100644
index 00000000..6ee73548
--- /dev/null
+++ b/tests/test_text_patch_eof_boundaries.py
@@ -0,0 +1,117 @@
+"""Unified patches cannot turn an EOF marker into an interior byte join."""
+
+import subprocess
+
+import pytest
+
+from msg.core.errors import Failure
+from msg.core.text_patch import apply_patch
+
+
+@pytest.mark.parametrize('rebase', [False, True])
+def test_new_eof_marker_rejects_remaining_source_lines_like_git(tmp_path, rebase):
+ source = 'a\nuntouched\n'
+ diff = '--- a/f\n+++ b/f\n@@ -1 +1 @@\n-a\n+b\n\\ No newline at end of file\n'
+ target = tmp_path / 'f'
+ target.write_bytes(source.encode())
+ result = subprocess.run(
+ ['git', 'apply', '-'], cwd=tmp_path, input=diff.encode(), capture_output=True
+ )
+ assert result.returncode != 0
+ assert target.read_bytes() == source.encode()
+ kwargs = {'base_source': source} if rebase else {}
+ with pytest.raises(Failure) as caught:
+ apply_patch(source, {'kind': 'unified', 'diff': diff}, **kwargs)
+ assert caught.value.code == 'invalid_unified_patch'
+
+
+def test_unterminated_diff_payload_requires_explicit_eof_marker_like_git(tmp_path):
+ source = 'a\n'
+ diff = '--- a/f\n+++ b/f\n@@ -1 +1 @@\n-a\n+b'
+ target = tmp_path / 'f'
+ target.write_bytes(source.encode())
+ result = subprocess.run(
+ ['git', 'apply', '-'], cwd=tmp_path, input=diff.encode(), capture_output=True
+ )
+ assert result.returncode != 0
+ assert target.read_bytes() == source.encode()
+ with pytest.raises(Failure) as caught:
+ apply_patch(source, {'kind': 'unified', 'diff': diff})
+ assert caught.value.code == 'invalid_unified_patch'
+
+
+@pytest.mark.parametrize('rebase', [False, True])
+def test_real_eof_marker_still_matches_git(tmp_path, rebase):
+ source = 'prefix\na\n'
+ diff = '--- a/f\n+++ b/f\n@@ -2 +2 @@\n-a\n+b\n\\ No newline at end of file\n'
+ target = tmp_path / 'f'
+ target.write_bytes(source.encode())
+ subprocess.run(
+ ['git', 'apply', '-'], cwd=tmp_path, input=diff.encode(), check=True, capture_output=True
+ )
+ kwargs = {'base_source': source} if rebase else {}
+ assert (
+ apply_patch(source, {'kind': 'unified', 'diff': diff}, **kwargs).encode()
+ == target.read_bytes()
+ )
+ assert target.read_bytes() == b'prefix\nb'
+
+
+def test_rebase_does_not_join_concurrently_appended_lines():
+ base = 'a\n'
+ current = base + 'concurrent\n'
+ diff = '@@ -1 +1 @@\n-a\n+b\n\\ No newline at end of file\n'
+ assert apply_patch(base, {'kind': 'unified', 'diff': diff}) == 'b'
+ with pytest.raises(Failure) as caught:
+ apply_patch(current, {'kind': 'unified', 'diff': diff}, base_source=base)
+ assert caught.value.code == 'invalid_unified_patch'
+
+
+@pytest.mark.asyncio
+async def test_signed_patch_failure_does_not_publish_revision(installed):
+ from test_service import call, register
+
+ from msg.core.codec import wire
+
+ app, _ = installed
+ key, owner, _ = await register(app, 'patch-eof-owner')
+ post = await call(
+ app,
+ 'content.post_create',
+ {'parent': '/main', 'body': 'a\nuntouched\n'},
+ key=key,
+ subject=owner,
+ )
+ assert post.status == 'ok', wire(post)
+ rid = post.resources[0].id
+
+ async def snapshot():
+ async with app.metadata.transaction(write=False) as tx:
+ resource = await tx.resource(rid)
+ return (
+ resource.generation,
+ resource.revision,
+ tx.one('SELECT COUNT(*) FROM revisions')[0],
+ tx.one('SELECT COUNT(*) FROM events')[0],
+ )
+
+ before = await snapshot()
+ failed = await call(
+ app,
+ 'content.text_patch',
+ {
+ 'id': rid,
+ 'base_revision': post.resources[0].revision,
+ 'base_generation': post.data['generation'],
+ 'patch': {
+ 'kind': 'unified',
+ 'diff': '@@ -1 +1 @@\n-a\n+b\n\\ No newline at end of file\n',
+ },
+ },
+ key=key,
+ subject=owner,
+ expected=((rid, post.data['generation']),),
+ contract_version=3,
+ )
+ assert failed.status == 'error' and failed.error.code == 'invalid_unified_patch', wire(failed)
+ assert await snapshot() == before
diff --git a/tests/test_webhook_delivery.py b/tests/test_webhook_delivery.py
index 37f7818a..6f1d9765 100644
--- a/tests/test_webhook_delivery.py
+++ b/tests/test_webhook_delivery.py
@@ -11,7 +11,13 @@
from msg.core.codec import b64, canonical, wire
from msg.core.errors import Failure
from msg.workers.effects import EffectWorker
-from msg.workers.webhook import PublicResolver, sign_delivery, validate_endpoint, verify_delivery
+from msg.workers.webhook import (
+ PublicResolver,
+ WebhookSender,
+ sign_delivery,
+ validate_endpoint,
+ verify_delivery,
+)
class Sender:
@@ -196,6 +202,8 @@ async def send(self, *args, **kwargs):
'https://[::1]/hook',
'https://10.1.2.3/hook',
'https://example.org:8443/hook',
+ 'https://example.org:0/hook',
+ 'https://example.org:000/hook',
'https://user:pass@example.org/hook',
],
)
@@ -204,6 +212,28 @@ def test_webhook_rejects_unsafe_endpoint(url):
validate_endpoint(url)
+@pytest.mark.parametrize('authority', ['example.org', 'example.org:443'])
+def test_webhook_accepts_only_default_or_explicit_https_port(authority):
+ assert validate_endpoint(f'https://{authority}/hook') == ('example.org', 443)
+
+
+@pytest.mark.asyncio
+async def test_webhook_zero_port_rejected_before_network_setup(monkeypatch):
+ def unexpected_connector(*args, **kwargs):
+ pytest.fail('invalid endpoint reached network setup')
+
+ monkeypatch.setattr('msg.workers.webhook.aiohttp.TCPConnector', unexpected_connector)
+ with pytest.raises(Failure, match='webhook_port_forbidden'):
+ await WebhookSender().send(
+ 'https://example.org:0/hook',
+ b'x' * 32,
+ b'{}',
+ timestamp='1',
+ event_id='e_a',
+ delivery_id='job_a',
+ )
+
+
def test_webhook_signature_covers_raw_body_and_timestamp():
secret = secrets.token_bytes(32)
timestamp = str(int(NOW.timestamp()))