fix(hosting): deploy 条目数与 preview 共享 128 上限

src/msg/extensions/hosting.py

@@ -15,6 +15,8 @@
 from msg.plugins.common import check_access,create_resource,revise_resource,resolve,output_for,assert_generation,new_id
 from msg.plugins.schemas import obj,STRING,IDENTIFIER,REF
 
+MAX_HOSTING_ENTRIES=128
+
 
 def path_name(value):
     require(isinstance(value,str) and 0<len(value)<=2048 and not value.startswith('/') and '\\' not in value and
@@ -41,6 +43,8 @@ async def deploy(ctx,request,tx):
         require(resource.type=='website' and resource.state=='active','not_a_website')
         await check_access(app,ctx,request,tx,resource.id,'write')
         await assert_generation(request,resource)
+        # A deployment must stay within what a preview could have shown first.
+        require(len(request.arguments['entries'])<=MAX_HOSTING_ENTRIES,'too_many_hosting_entries')
         sources=[];names=set()
         for item in request.arguments['entries']:
             path=path_name(item['path'])
@@ -75,7 +79,7 @@ async def preview(ctx,request,tx):
         await assert_generation(request,website)
         # Keep the published v1 wire schema immutable; bound work before any
         # source traversal or materialization, just like runtime capacity checks.
-        require(len(request.arguments['entries'])<=128,'too_many_preview_entries')
+        require(len(request.arguments['entries'])<=MAX_HOSTING_ENTRIES,'too_many_preview_entries')
         sources=[];names=set()
         for item in request.arguments['entries']:
             path=path_name(item['path'])

tests/test_hosting_same_origin.py

@@ -80,6 +80,48 @@ async def test_same_domain_hosted_html_head_304_range_raw_and_no_write_route(ins
         assert forbidden.status_code==405
 
 
+@pytest.mark.asyncio
+async def test_deploy_entry_bound_matches_preview_before_any_materialization(installed,monkeypatch):
+    app, _ = installed
+    key,user,_=await register(app,'web-bounded')
+    site=await call(app,'hosting.create',{'parent':'/@web-bounded','name':'web'},
+                    key=key,subject=user)
+    source=await call(app,'content.file_put',{'parent':'/@web-bounded/files',
+        'name':'page.html','data':b64(b'<p>bounded</p>'),'media_type':'text/html'},
+        key=key,subject=user)
+    assert site.status==source.status=='ok'
+    site_id=site.resources[0].id
+    arguments={'id':site_id,'entries':[{'path':f'{i}.html','source':wire(source.resources[0])}
+                                       for i in range(129)]}
+    spec=app.registry.operation('hosting.deploy',1)
+    assert 'maxItems' not in app.registry.schema(spec.input_schema)['properties']['entries']
+    app.registry.validate(spec.input_schema,arguments)
+    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'))
+    async def no_materialization(*args,**kwargs):
+        pytest.fail('oversized deploy touched the content store')
+    with monkeypatch.context() as patch:
+        patch.setattr(app.contents,'read_bytes',no_materialization)
+        patch.setattr(app.contents,'put_bytes',no_materialization)
+        denied=await call(app,'hosting.deploy',arguments,key=key,subject=user,
+                          expected=((site_id,site.data['generation']),))
+    assert denied.status=='error' and denied.error.code=='too_many_hosting_entries',wire(denied)
+    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
+        assert (await tx.resource(site_id)).revision is None
+    allowed=await call(app,'hosting.deploy',{**arguments,'entries':arguments['entries'][:128]},
+                       key=key,subject=user,expected=((site_id,site.data['generation']),))
+    assert allowed.status=='ok' and allowed.data['files']==128,wire(allowed)
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=create_app(app)),
+                                 base_url='http://testserver') as http:
+        last=await http.get('/@web-bounded/web/127.html')
+        assert last.status_code==200 and last.content==b'<p>bounded</p>'
+        isolated(last)
+        assert (await http.get('/@web-bounded/web/128.html')).status_code==404
+
+
 @pytest.mark.asyncio
 async def test_preview_history_and_atomic_activation_share_same_csp(installed):
     app, _ = installed