fix: toolsrv 9P server now handles BlockOnce/Stream reads correctly
- Added BlockingReadMode() and StreamMode() handling to toolsrv's handleRead() - Made NextPending() context-aware in both toolsrv and olliesrv bypass packages - This fixes the bypass approval flow via 9P - reads now properly block until a request is available and respect context cancellation
This commit is contained in:
parent
d59f49ee54
commit
46bd0dd999
|
|
@ -1,6 +1,7 @@
|
|||
package bypass
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
|
@ -84,8 +85,14 @@ func (b *Broker) Pending() []*Request {
|
|||
}
|
||||
|
||||
// NextPending blocks until a bypass request is available and returns it.
|
||||
func (b *Broker) NextPending() *Request {
|
||||
return <-b.pendingCh
|
||||
// Returns nil if context is cancelled before a request arrives.
|
||||
func (b *Broker) NextPending(ctx context.Context) *Request {
|
||||
select {
|
||||
case req := <-b.pendingCh:
|
||||
return req
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// PendingByID returns a specific pending request.
|
||||
|
|
|
|||
|
|
@ -171,13 +171,13 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl {
|
|||
virtfs.Doc("Bypass broker for sandboxed command approval"),
|
||||
virtfs.FileNode("pending", 0444,
|
||||
virtfs.Doc("Blocks until bypass request; returns JSON {id, cmd, cwd, env}"),
|
||||
virtfs.BlockOnce(func(_ context.Context, _ string) ([]byte, string, error) {
|
||||
virtfs.BlockOnce(func(ctx context.Context, _ string) ([]byte, string, error) {
|
||||
if broker == nil {
|
||||
return nil, "", errNoBypass
|
||||
}
|
||||
req := broker.NextPending()
|
||||
req := broker.NextPending(ctx)
|
||||
if req == nil {
|
||||
return nil, "", fmt.Errorf("no pending request")
|
||||
return nil, "", fmt.Errorf("context cancelled")
|
||||
}
|
||||
data, err := json.Marshal(req)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -75,8 +75,14 @@ func Submit(ctx context.Context, cmd, cwd string, env map[string]string) (bool,
|
|||
|
||||
// NextPending blocks until a request is available and returns it.
|
||||
// Used by the 9P pending file to expose requests to the approver.
|
||||
func NextPending() *Request {
|
||||
return <-pending
|
||||
// Returns nil if context is cancelled before a request arrives.
|
||||
func NextPending(ctx context.Context) *Request {
|
||||
select {
|
||||
case req := <-pending:
|
||||
return req
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// Resolve completes a pending request with the given decision.
|
||||
|
|
|
|||
|
|
@ -134,8 +134,11 @@ func Spec(srv *Server) virtfs.FsNodeDecl {
|
|||
virtfs.DirNode("bypass",
|
||||
virtfs.FileNode("pending", 0444,
|
||||
virtfs.Doc("Blocks until bypass request; returns JSON {id, cmd, cwd, env}"),
|
||||
virtfs.BlockOnce(func(_ context.Context, _ string) ([]byte, string, error) {
|
||||
req := bypass.NextPending()
|
||||
virtfs.BlockOnce(func(ctx context.Context, _ string) ([]byte, string, error) {
|
||||
req := bypass.NextPending(ctx)
|
||||
if req == nil {
|
||||
return nil, "", fmt.Errorf("context cancelled")
|
||||
}
|
||||
data, err := json.Marshal(req)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ type fid struct {
|
|||
entry virtfs.File
|
||||
readBuf []byte // buffer for read data
|
||||
readDone bool // true if file has been read
|
||||
waitBase string // base string for blocking/stream reads
|
||||
dirCache []byte
|
||||
}
|
||||
|
||||
|
|
@ -375,7 +376,52 @@ func handleRead(cs *connState, fc *plan9.Fcall, tree *virtfs.Tree) *plan9.Fcall
|
|||
return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Data: nil}
|
||||
}
|
||||
|
||||
// Buffer read data if not already done
|
||||
// Handle BlockOnce mode: block until content is available, then cache it
|
||||
if f.entry.BlockingReadMode() {
|
||||
// Serve from cache if we already have content (multi-Tread paging)
|
||||
if f.readBuf != nil {
|
||||
if int(fc.Offset) >= len(f.readBuf) {
|
||||
return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: 0}
|
||||
}
|
||||
return readSlice(fc, f.readBuf)
|
||||
}
|
||||
// First Tread: block for content
|
||||
content, nextBase, err := f.entry.BlockingRead(cs.ctx, f.waitBase)
|
||||
if err != nil {
|
||||
return errFcall(fc, err.Error())
|
||||
}
|
||||
if nextBase != "" {
|
||||
cs.mu.Lock()
|
||||
if ff, ok := cs.fids[fc.Fid]; ok {
|
||||
ff.waitBase = nextBase
|
||||
}
|
||||
cs.mu.Unlock()
|
||||
}
|
||||
f.readBuf = content
|
||||
return readSlice(fc, content)
|
||||
}
|
||||
|
||||
// Handle Stream mode: block indefinitely, stream data incrementally
|
||||
if f.entry.StreamMode() {
|
||||
content, nextBase, err := f.entry.BlockingRead(cs.ctx, f.waitBase)
|
||||
if err != nil {
|
||||
return errFcall(fc, err.Error())
|
||||
}
|
||||
if nextBase != "" {
|
||||
cs.mu.Lock()
|
||||
if ff, ok := cs.fids[fc.Fid]; ok {
|
||||
ff.waitBase = nextBase
|
||||
}
|
||||
cs.mu.Unlock()
|
||||
}
|
||||
count := int(fc.Count)
|
||||
if count > len(content) {
|
||||
count = len(content)
|
||||
}
|
||||
return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: uint32(count), Data: content[:count]}
|
||||
}
|
||||
|
||||
// Buffer read data if not already done (normal non-blocking read)
|
||||
if f.readBuf == nil {
|
||||
data, err := f.entry.Read()
|
||||
if err != nil && err != io.EOF {
|
||||
|
|
@ -399,6 +445,20 @@ func handleRead(cs *connState, fc *plan9.Fcall, tree *virtfs.Tree) *plan9.Fcall
|
|||
return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Data: f.readBuf[offset:end]}
|
||||
}
|
||||
|
||||
// readSlice returns a slice of content based on offset/count from fc.
|
||||
func readSlice(fc *plan9.Fcall, content []byte) *plan9.Fcall {
|
||||
var data []byte
|
||||
off := int(fc.Offset)
|
||||
if off < len(content) {
|
||||
end := off + int(fc.Count)
|
||||
if end > len(content) {
|
||||
end = len(content)
|
||||
}
|
||||
data = content[off:end]
|
||||
}
|
||||
return &plan9.Fcall{Type: plan9.Rread, Tag: fc.Tag, Count: uint32(len(data)), Data: data}
|
||||
}
|
||||
|
||||
func handleDirRead(cs *connState, f *fid, fc *plan9.Fcall, tree *virtfs.Tree) *plan9.Fcall {
|
||||
if fc.Offset == 0 || f.dirCache == nil {
|
||||
path := f.path
|
||||
|
|
|
|||
Loading…
Reference in New Issue