|
@@ -14,21 +14,23 @@ const hash = (value: string) => createHash('sha256').update(value).digest('hex')
|
|
|
const input = z.object({ name:z.string().trim().min(1).max(100),sourceNodeId:z.string().min(1).max(100).default('local'),
|
|
const input = z.object({ name:z.string().trim().min(1).max(100),sourceNodeId:z.string().min(1).max(100).default('local'),
|
|
|
allowedProfiles:z.array(z.string().min(1).max(100)).min(1).max(256) }).strict()
|
|
allowedProfiles:z.array(z.string().min(1).max(100)).min(1).max(256) }).strict()
|
|
|
interface Pending { command:RunnerCommand; resolve(value:any):void; reject(error:Error):void; timer:ReturnType<typeof setTimeout>; nextDelivery:number; valid():boolean }
|
|
interface Pending { command:RunnerCommand; resolve(value:any):void; reject(error:Error):void; timer:ReturnType<typeof setTimeout>; nextDelivery:number; valid():boolean }
|
|
|
-interface GatewayConnection { runnerId:string; valid():boolean; onEvent(frame:GatewayFrame):void; onDisconnect():void }
|
|
|
|
|
|
|
+interface GatewayConnection { runnerId:string;cleanupOnly?:boolean;publishArtifact?:(name:string,bytes:Buffer)=>Promise<unknown>; valid():boolean; onEvent(frame:GatewayFrame):void; onDisconnect():void }
|
|
|
interface Lease { runnerId:string; input:LeaseInput; calls:Map<string,{fingerprint:string;result:Promise<unknown>}> }
|
|
interface Lease { runnerId:string; input:LeaseInput; calls:Map<string,{fingerprint:string;result:Promise<unknown>}> }
|
|
|
|
|
|
|
|
/** Outbound-only machine transport. Browser auth and machine auth remain separate. */
|
|
/** Outbound-only machine transport. Browser auth and machine auth remain separate. */
|
|
|
export class RunnerHub {
|
|
export class RunnerHub {
|
|
|
readonly epoch=randomUUID()
|
|
readonly epoch=randomUUID()
|
|
|
- private online = new Map<string,{instance:string;seen:number;epoch:string;wake?:()=>void}>()
|
|
|
|
|
|
|
+ private online = new Map<string,{instance:string;seen:number;epoch:string;features:string[];wake?:()=>void}>()
|
|
|
private retired = new Map<string,Set<string>>()
|
|
private retired = new Map<string,Set<string>>()
|
|
|
private pending = new Map<string,Map<string,Pending>>()
|
|
private pending = new Map<string,Map<string,Pending>>()
|
|
|
private connections = new Map<string,GatewayConnection>()
|
|
private connections = new Map<string,GatewayConnection>()
|
|
|
|
|
+ private artifacts=new Map<string,{runnerId:string;connectionId:string;name:string;size:number;digest:string;chunks:Buffer[];bytes:number;next:number;result?:unknown}>()
|
|
|
private leases = new Map<string,Lease>()
|
|
private leases = new Map<string,Lease>()
|
|
|
private watchdog: ReturnType<typeof setInterval>
|
|
private watchdog: ReturnType<typeof setInterval>
|
|
|
constructor(readonly store:WorkspaceStore,readonly auth:LocalAuthStore,readonly local:GatewayTarget) {
|
|
constructor(readonly store:WorkspaceStore,readonly auth:LocalAuthStore,readonly local:GatewayTarget) {
|
|
|
this.watchdog=setInterval(()=>{
|
|
this.watchdog=setInterval(()=>{
|
|
|
for(const [id,state] of this.online) if(Date.now()-state.seen>35_000)this.disconnect(id)
|
|
for(const [id,state] of this.online) if(Date.now()-state.seen>35_000)this.disconnect(id)
|
|
|
|
|
+ for(const [id,upload] of this.artifacts)if(!this.connections.has(upload.connectionId))this.artifacts.delete(id)
|
|
|
},5000);this.watchdog.unref()
|
|
},5000);this.watchdog.unref()
|
|
|
}
|
|
}
|
|
|
records():RunnerRecord[] {return this.store.list('_system','runner')}
|
|
records():RunnerRecord[] {return this.store.list('_system','runner')}
|
|
@@ -46,7 +48,7 @@ export class RunnerHub {
|
|
|
}
|
|
}
|
|
|
summary(record:RunnerRecord) {
|
|
summary(record:RunnerRecord) {
|
|
|
const {tokenHash:_secret,sourceOwner:_owner,...publicRecord}=record
|
|
const {tokenHash:_secret,sourceOwner:_owner,...publicRecord}=record
|
|
|
- return {...publicRecord,online:!!this.online.get(record.id)&&Date.now()-this.online.get(record.id)!.seen<35_000}
|
|
|
|
|
|
|
+ return {...publicRecord,features:this.online.get(record.id)?.features??[],online:!!this.online.get(record.id)&&Date.now()-this.online.get(record.id)!.seen<35_000}
|
|
|
}
|
|
}
|
|
|
remove(id:string) {
|
|
remove(id:string) {
|
|
|
const record=this.store.require<RunnerRecord>('_system','runner',id)
|
|
const record=this.store.require<RunnerRecord>('_system','runner',id)
|
|
@@ -72,18 +74,20 @@ export class RunnerHub {
|
|
|
queue.set(command.id,{command,resolve,reject,timer,nextDelivery:0,valid});this.online.get(id)?.wake?.()
|
|
queue.set(command.id,{command,resolve,reject,timer,nextDelivery:0,valid});this.online.get(id)?.wake?.()
|
|
|
})
|
|
})
|
|
|
}
|
|
}
|
|
|
- target(owner:string,sourceNodeId:string):GatewayTarget|undefined {
|
|
|
|
|
|
|
+ target(owner:string,sourceNodeId:string,computer?:import('../runner/worker/gateway.js').ComputerTarget):GatewayTarget|undefined {
|
|
|
const record=this.records().find(r=>r.enabled&&r.sourceNodeId===sourceNodeId&&(r.sourceOwner==='_system'||r.sourceOwner===owner))
|
|
const record=this.records().find(r=>r.enabled&&r.sourceNodeId===sourceNodeId&&(r.sourceOwner==='_system'||r.sourceOwner===owner))
|
|
|
if(!record)return undefined
|
|
if(!record)return undefined
|
|
|
const requireProfile=(profile:string)=>{
|
|
const requireProfile=(profile:string)=>{
|
|
|
if(!this.auth.canUseSource(owner,sourceNodeId,profile)||!record.allowedProfiles.includes(profile))throw new HttpError(403,'执行节点未授权这个基础 Profile','runner_profile_forbidden')
|
|
if(!this.auth.canUseSource(owner,sourceNodeId,profile)||!record.allowedProfiles.includes(profile))throw new HttpError(403,'执行节点未授权这个基础 Profile','runner_profile_forbidden')
|
|
|
}
|
|
}
|
|
|
|
|
+ const requireComputer=()=>{if(computer&&!this.online.get(record.id)?.features.includes('computer-worker-v1'))throw new HttpError(409,'执行节点未提供隔离 Worker 能力,请检查配置或更新 Runner','computer_unavailable')}
|
|
|
const requestHTTP=async(path:string,options:UpstreamRequestOptions={}):Promise<UpstreamResponse>=>{
|
|
const requestHTTP=async(path:string,options:UpstreamRequestOptions={}):Promise<UpstreamResponse>=>{
|
|
|
|
|
+ requireComputer()
|
|
|
const profile=options.search?.get('profile')??'default'
|
|
const profile=options.search?.get('profile')??'default'
|
|
|
if(path!=='/api/profiles'&&path!=='/api/status')requireProfile(profile)
|
|
if(path!=='/api/profiles'&&path!=='/api/status')requireProfile(profile)
|
|
|
const version=this.auth.pushAuthorizationVersion(owner)
|
|
const version=this.auth.pushAuthorizationVersion(owner)
|
|
|
const valid=()=>this.auth.isUserActive(owner)&&this.auth.pushAuthorizationVersion(owner)===version&&this.store.get<RunnerRecord>('_system','runner',record.id)?.enabled===true
|
|
const valid=()=>this.auth.isUserActive(owner)&&this.auth.pushAuthorizationVersion(owner)===version&&this.store.get<RunnerRecord>('_system','runner',record.id)?.enabled===true
|
|
|
- const result=await this.request(record.id,'http',{path,method:options.method??'GET',search:options.search?.toString(),body:options.body},valid)
|
|
|
|
|
|
|
+ const result=await this.request(record.id,'http',{path,method:options.method??'GET',search:options.search?.toString(),body:options.body,computer},valid)
|
|
|
if(!valid())throw new HttpError(403,'读取权限已失效','runner_authorization_revoked')
|
|
if(!valid())throw new HttpError(403,'读取权限已失效','runner_authorization_revoked')
|
|
|
if(path==='/api/profiles') {
|
|
if(path==='/api/profiles') {
|
|
|
const parsed=JSON.parse(Buffer.from(result.body,'base64').toString())
|
|
const parsed=JSON.parse(Buffer.from(result.body,'base64').toString())
|
|
@@ -93,17 +97,21 @@ export class RunnerHub {
|
|
|
return {status:result.status,headers:new Headers(result.headers),body:Buffer.from(result.body,'base64')}
|
|
return {status:result.status,headers:new Headers(result.headers),body:Buffer.from(result.body,'base64')}
|
|
|
}
|
|
}
|
|
|
return {url:this.local.url,client:this.local.client,session:{request:requestHTTP,webSocketCredential:async()=>{throw new Error('Runner uses its own authenticated gateway')}},
|
|
return {url:this.local.url,client:this.local.client,session:{request:requestHTTP,webSocketCredential:async()=>{throw new Error('Runner uses its own authenticated gateway')}},
|
|
|
- runner:{id:record.id,
|
|
|
|
|
- open:async(onEvent,onDisconnect)=>{
|
|
|
|
|
|
|
+ runner:{id:record.id,computer:!!computer,
|
|
|
|
|
+ open:async(onEvent,onDisconnect,scope)=>{
|
|
|
|
|
+ requireComputer()
|
|
|
|
|
+ if(computer&&!scope)throw new HttpError(403,'隔离执行缺少任务授权','computer_scope_required')
|
|
|
const version=this.auth.pushAuthorizationVersion(owner)
|
|
const version=this.auth.pushAuthorizationVersion(owner)
|
|
|
- const valid=()=>this.auth.isUserActive(owner)&&this.auth.pushAuthorizationVersion(owner)===version&&this.store.get<RunnerRecord>('_system','runner',record.id)?.enabled===true
|
|
|
|
|
- const id=randomUUID();this.connections.set(id,{runnerId:record.id,valid,onEvent,onDisconnect})
|
|
|
|
|
- try {await this.request(record.id,'gateway.open',{connectionId:id})} catch(error){this.connections.delete(id);void this.request(record.id,'gateway.close',{connectionId:id}).catch(()=>{});throw error}
|
|
|
|
|
|
|
+ const valid=()=>{try{scope?.authorize();return (scope?.cleanupOnly||(this.auth.isUserActive(owner)&&this.auth.pushAuthorizationVersion(owner)===version))&&this.store.get<RunnerRecord>('_system','runner',record.id)?.enabled===true}catch{return false}}
|
|
|
|
|
+ const id=randomUUID();this.connections.set(id,{runnerId:record.id,valid,onEvent,onDisconnect,publishArtifact:scope?.publishArtifact,cleanupOnly:scope?.cleanupOnly})
|
|
|
|
|
+ try {await this.request(record.id,'gateway.open',{connectionId:id,computer,workId:scope?.workId,cleanupOnly:scope?.cleanupOnly})} catch(error){this.connections.delete(id);void this.request(record.id,'gateway.close',{connectionId:id}).catch(()=>{});throw error}
|
|
|
let closed=false
|
|
let closed=false
|
|
|
return {rpc:async(method,params)=>{
|
|
return {rpc:async(method,params)=>{
|
|
|
if(closed)throw new Error('Runner gateway closed')
|
|
if(closed)throw new Error('Runner gateway closed')
|
|
|
- if(!valid())throw new HttpError(403,'本轮执行授权已失效','runner_command_not_admitted')
|
|
|
|
|
- if(method==='session.create'||method==='session.resume')requireProfile(String(params.profile??'default'))
|
|
|
|
|
|
|
+ const cleanup=['session.interrupt','session.close'].includes(method)
|
|
|
|
|
+ if(scope?.cleanupOnly&&(!['session.resume','session.interrupt','session.close'].includes(method)||(method==='session.resume'&¶ms.session_id!==scope.sessionId)))throw new HttpError(403,'清理通道只能操作原绑定会话','runner_cleanup_forbidden')
|
|
|
|
|
+ if(!cleanup&&!valid())throw new HttpError(403,'本轮执行授权已失效','runner_command_not_admitted')
|
|
|
|
|
+ if(!scope?.cleanupOnly&&(method==='session.create'||method==='session.resume'))requireProfile(String(params.profile??'default'))
|
|
|
return this.request(record.id,'gateway.rpc',{connectionId:id,method,params})
|
|
return this.request(record.id,'gateway.rpc',{connectionId:id,method,params})
|
|
|
},close:()=>{
|
|
},close:()=>{
|
|
|
closed=true;this.connections.delete(id)
|
|
closed=true;this.connections.delete(id)
|
|
@@ -112,7 +120,7 @@ export class RunnerHub {
|
|
|
}}
|
|
}}
|
|
|
},
|
|
},
|
|
|
lease:async(options)=>{
|
|
lease:async(options)=>{
|
|
|
- requireProfile(options.profile);options.assertActive()
|
|
|
|
|
|
|
+ requireComputer();requireProfile(options.profile);options.assertActive()
|
|
|
const id=randomUUID();this.leases.set(id,{runnerId:record.id,input:options,calls:new Map()})
|
|
const id=randomUUID();this.leases.set(id,{runnerId:record.id,input:options,calls:new Map()})
|
|
|
let disposed=false
|
|
let disposed=false
|
|
|
const dispose=async()=>{if(disposed)return;disposed=true;this.leases.delete(id);options.signal.removeEventListener('abort',abort);await this.request(record.id,'lease.close',{leaseId:id}).catch(()=>{})}
|
|
const dispose=async()=>{if(disposed)return;disposed=true;this.leases.delete(id);options.signal.removeEventListener('abort',abort);await this.request(record.id,'lease.close',{leaseId:id}).catch(()=>{})}
|
|
@@ -135,7 +143,7 @@ export class RunnerHub {
|
|
|
}
|
|
}
|
|
|
middleware():Koa.Middleware {
|
|
middleware():Koa.Middleware {
|
|
|
return async(ctx,next)=>{
|
|
return async(ctx,next)=>{
|
|
|
- const match=/^\/api\/runner\/v1\/([0-9a-f-]{36})\/(poll|admit|result|event|tool)$/.exec(ctx.path)
|
|
|
|
|
|
|
+ const match=/^\/api\/runner\/v1\/([0-9a-f-]{36})\/(poll|admit|result|event|tool|check|artifact)$/.exec(ctx.path)
|
|
|
if(!match)return next()
|
|
if(!match)return next()
|
|
|
const record=this.store.get<RunnerRecord>('_system','runner',match[1]!),token=ctx.get('authorization').replace(/^Bearer /,'')
|
|
const record=this.store.get<RunnerRecord>('_system','runner',match[1]!),token=ctx.get('authorization').replace(/^Bearer /,'')
|
|
|
if(ctx.get('origin')||!record?.enabled||!ctx.get('authorization').startsWith('Bearer ')||!timingSafeEqual(Buffer.from(record.tokenHash),Buffer.from(hash(token))))
|
|
if(ctx.get('origin')||!record?.enabled||!ctx.get('authorization').startsWith('Bearer ')||!timingSafeEqual(Buffer.from(record.tokenHash),Buffer.from(hash(token))))
|
|
@@ -152,7 +160,7 @@ export class RunnerHub {
|
|
|
retired.add(previous.instance);this.retired.set(record.id,retired);this.disconnect(record.id);previous=undefined
|
|
retired.add(previous.instance);this.retired.set(record.id,retired);this.disconnect(record.id);previous=undefined
|
|
|
}
|
|
}
|
|
|
if(match[2]!=='poll'&&ctx.get('x-runner-epoch')!==previous?.epoch)throw new HttpError(409,'执行连接代次已改变','runner_epoch_changed')
|
|
if(match[2]!=='poll'&&ctx.get('x-runner-epoch')!==previous?.epoch)throw new HttpError(409,'执行连接代次已改变','runner_epoch_changed')
|
|
|
- const state=this.online.get(record.id)??{instance,seen:Date.now(),epoch:`${this.epoch}:${randomUUID()}`};state.seen=Date.now();this.online.set(record.id,state)
|
|
|
|
|
|
|
+ const state=this.online.get(record.id)??{instance,seen:Date.now(),features:[],epoch:`${this.epoch}:${randomUUID()}`};state.seen=Date.now();if(match[2]==='poll')state.features=ctx.get('x-runner-features').split(',').filter(value=>['computer-worker-v1','artifact-chunks-v1'].includes(value));this.online.set(record.id,state)
|
|
|
ctx.set('Cache-Control','no-store')
|
|
ctx.set('Cache-Control','no-store')
|
|
|
if(match[2]==='poll') {
|
|
if(match[2]==='poll') {
|
|
|
if(ctx.method!=='GET')throw new HttpError(405,'仅允许 GET','method_not_allowed')
|
|
if(ctx.method!=='GET')throw new HttpError(405,'仅允许 GET','method_not_allowed')
|
|
@@ -173,10 +181,40 @@ export class RunnerHub {
|
|
|
let body:any
|
|
let body:any
|
|
|
try {body=JSON.parse(Buffer.concat(chunks).toString())}catch{throw new HttpError(400,'JSON 无效','invalid_json')}
|
|
try {body=JSON.parse(Buffer.concat(chunks).toString())}catch{throw new HttpError(400,'JSON 无效','invalid_json')}
|
|
|
if(!body||typeof body!=='object'||Array.isArray(body))throw new HttpError(400,'请求必须是对象','invalid_json')
|
|
if(!body||typeof body!=='object'||Array.isArray(body))throw new HttpError(400,'请求必须是对象','invalid_json')
|
|
|
|
|
+ if(match[2]==='artifact'){
|
|
|
|
|
+ const value=parse(z.object({connectionId:z.string().uuid(),id:z.string().uuid(),phase:z.enum(['begin','append','finish']),name:z.string().min(1).max(240).optional(),size:z.number().int().min(0).max(25*1024*1024).optional(),digest:z.string().regex(/^[a-f0-9]{64}$/).optional(),index:z.number().int().min(0).optional(),data:z.string().max(1024*1024).optional()}).strict(),body)
|
|
|
|
|
+ const connection=this.connections.get(value.connectionId)
|
|
|
|
|
+ if(connection?.runnerId!==record.id||!connection.valid()||!connection.publishArtifact)throw new HttpError(403,'产物不属于当前有效任务','artifact_forbidden')
|
|
|
|
|
+ let upload=this.artifacts.get(value.id)
|
|
|
|
|
+ if(value.phase==='begin'){
|
|
|
|
|
+ if(!value.name||value.size===undefined||!value.digest)throw new HttpError(400,'产物信息不完整','artifact_invalid')
|
|
|
|
|
+ if(upload&&(upload.connectionId!==value.connectionId||upload.name!==value.name||upload.size!==value.size||upload.digest!==value.digest))throw new HttpError(409,'产物编号已用于其他内容','idempotency_conflict')
|
|
|
|
|
+ if(!upload){
|
|
|
|
|
+ const current=[...this.artifacts.values()]
|
|
|
|
|
+ if(current.filter(item=>item.connectionId===value.connectionId).length>=8||current.filter(item=>item.runnerId===record.id&&item.result===undefined).reduce((sum,item)=>sum+item.size,0)+value.size>64*1024*1024)throw new HttpError(429,'待传产物过多','artifact_limit')
|
|
|
|
|
+ upload={runnerId:record.id,connectionId:value.connectionId,name:value.name,size:value.size,digest:value.digest,chunks:[],bytes:0,next:0};this.artifacts.set(value.id,upload)
|
|
|
|
|
+ }
|
|
|
|
|
+ }else{
|
|
|
|
|
+ if(!upload||upload.connectionId!==value.connectionId||upload.runnerId!==record.id)throw new HttpError(404,'产物传输不存在','artifact_missing')
|
|
|
|
|
+ if(value.phase==='append'){
|
|
|
|
|
+ if(value.index===undefined||value.data===undefined||upload.result!==undefined)throw new HttpError(400,'产物分块无效','artifact_invalid')
|
|
|
|
|
+ const bytes=Buffer.from(value.data,'base64')
|
|
|
|
|
+ if(value.index<upload.next){if(!upload.chunks[value.index]?.equals(bytes))throw new HttpError(409,'重复分块内容不一致','idempotency_conflict')}
|
|
|
|
|
+ else{if(value.index!==upload.next||upload.bytes+bytes.length>upload.size)throw new HttpError(409,'分块顺序或大小无效','artifact_invalid');upload.chunks.push(bytes);upload.bytes+=bytes.length;upload.next++}
|
|
|
|
|
+ }else if(upload.result===undefined){
|
|
|
|
|
+ if(upload.bytes!==upload.size)throw new HttpError(409,'产物尚未传输完整','artifact_incomplete')
|
|
|
|
|
+ const bytes=Buffer.concat(upload.chunks)
|
|
|
|
|
+ if(createHash('sha256').update(bytes).digest('hex')!==upload.digest)throw new HttpError(409,'产物校验失败','artifact_digest_mismatch')
|
|
|
|
|
+ upload.result=await connection.publishArtifact(upload.name,bytes);upload.chunks=[]
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ ctx.body={ok:true,next:upload!.next,result:upload!.result};return
|
|
|
|
|
+ }
|
|
|
|
|
+ if(match[2]==='check'){const connection=this.connections.get(body.connectionId);ctx.body={allowed:connection?.runnerId===record.id&&connection.valid()};return}
|
|
|
if(match[2]==='admit') {
|
|
if(match[2]==='admit') {
|
|
|
const pending=this.pending.get(record.id)?.get(body.id),command=pending?.command
|
|
const pending=this.pending.get(record.id)?.get(body.id),command=pending?.command
|
|
|
let allowed=!!command&&command.expiresAt>Date.now()&&pending!.valid()
|
|
let allowed=!!command&&command.expiresAt>Date.now()&&pending!.valid()
|
|
|
- if(command?.kind==='gateway.rpc'||command?.kind==='gateway.open')allowed&&=this.connections.get(String(command.payload.connectionId))?.valid()===true
|
|
|
|
|
|
|
+ if(command?.kind==='gateway.rpc'||command?.kind==='gateway.open'){const connection=this.connections.get(String(command.payload.connectionId));allowed&&=connection?.runnerId===record.id&&(command.kind==='gateway.rpc'&&['session.interrupt','session.close'].includes(String(command.payload.method))||connection.valid())}
|
|
|
if(command?.kind==='lease.create'||command?.kind==='lease.bind') {
|
|
if(command?.kind==='lease.create'||command?.kind==='lease.bind') {
|
|
|
const lease=this.leases.get(String(command.payload.leaseId))
|
|
const lease=this.leases.get(String(command.payload.leaseId))
|
|
|
try {if(lease?.runnerId!==record.id)allowed=false;else lease.input.assertActive()}catch{allowed=false}
|
|
try {if(lease?.runnerId!==record.id)allowed=false;else lease.input.assertActive()}catch{allowed=false}
|