|
| 1 | +package engine |
| 2 | + |
| 3 | +import ( |
| 4 | + "io" |
| 5 | + "strings" |
| 6 | + "sync" |
| 7 | +) |
| 8 | + |
| 9 | +// PassthroughPrefix is prepended to every line of engine subprocess output that |
| 10 | +// the detector forwards to its own standard error. The engine analyzes |
| 11 | +// attacker-controlled artifacts, so its output is untrusted: without a frame, |
| 12 | +// model-authored text containing a newline could impersonate a detector |
| 13 | +// diagnostic (a THREAT_DETECTION_* marker) or a host workflow command. The |
| 14 | +// prefix makes forwarded bytes distinguishable — for humans reading the job log |
| 15 | +// and for tooling that scans it — while preserving real-time streaming. |
| 16 | +// |
| 17 | +// Consumers that scan the captured log for detector-attested markers MUST |
| 18 | +// ignore lines carrying this prefix. |
| 19 | +const PassthroughPrefix = "[engine] " |
| 20 | + |
| 21 | +// maxPassthroughLineBytes bounds how much engine output is buffered while |
| 22 | +// waiting for a line terminator. An engine that never emits a newline would |
| 23 | +// otherwise grow the buffer without limit; past this many bytes the pending |
| 24 | +// content is flushed as its own framed line. |
| 25 | +const maxPassthroughLineBytes = 8192 |
| 26 | + |
| 27 | +// passthroughFramer forwards engine subprocess output to a destination writer, |
| 28 | +// one framed line at a time. A single framer serves both the stdout and stderr |
| 29 | +// streams of one subprocess: it owns the lock that keeps their interleaved |
| 30 | +// writes from splicing into each other's lines. |
| 31 | +type passthroughFramer struct { |
| 32 | + mu sync.Mutex |
| 33 | + dst io.Writer |
| 34 | + writers []*passthroughWriter |
| 35 | +} |
| 36 | + |
| 37 | +func newPassthroughFramer(dst io.Writer) *passthroughFramer { |
| 38 | + return &passthroughFramer{dst: dst} |
| 39 | +} |
| 40 | + |
| 41 | +// writer returns a new stream writer bound to this framer. It must be called |
| 42 | +// before the subprocess starts, since the writer list is not itself guarded. |
| 43 | +func (f *passthroughFramer) writer() io.Writer { |
| 44 | + w := &passthroughWriter{framer: f} |
| 45 | + f.writers = append(f.writers, w) |
| 46 | + return w |
| 47 | +} |
| 48 | + |
| 49 | +// Close flushes any partial line each stream left behind — output that ended |
| 50 | +// without a trailing newline, which would otherwise be lost. |
| 51 | +func (f *passthroughFramer) Close() { |
| 52 | + f.mu.Lock() |
| 53 | + defer f.mu.Unlock() |
| 54 | + for _, w := range f.writers { |
| 55 | + w.flushLocked() |
| 56 | + } |
| 57 | +} |
| 58 | + |
| 59 | +// passthroughWriter accumulates one stream's bytes and emits a framed line each |
| 60 | +// time a terminator is seen. |
| 61 | +type passthroughWriter struct { |
| 62 | + framer *passthroughFramer |
| 63 | + buf []byte |
| 64 | + pendingCR bool |
| 65 | +} |
| 66 | + |
| 67 | +// Write never reports an error: forwarding is a diagnostic convenience, and |
| 68 | +// failing the subprocess because the job log could not be written would turn a |
| 69 | +// cosmetic problem into a detection outage. |
| 70 | +func (w *passthroughWriter) Write(p []byte) (int, error) { |
| 71 | + f := w.framer |
| 72 | + f.mu.Lock() |
| 73 | + defer f.mu.Unlock() |
| 74 | + for _, b := range p { |
| 75 | + if w.pendingCR { |
| 76 | + w.pendingCR = false |
| 77 | + // A CR already terminated the line; swallow the LF of a CRLF pair |
| 78 | + // rather than emitting an empty line for it. |
| 79 | + if b == '\n' { |
| 80 | + continue |
| 81 | + } |
| 82 | + } |
| 83 | + switch b { |
| 84 | + case '\n': |
| 85 | + w.emitLocked() |
| 86 | + case '\r': |
| 87 | + // A bare CR is treated as a terminator too: the Actions runner |
| 88 | + // splits process output on CR as well as LF, so leaving one inline |
| 89 | + // would let engine output start a line the runner then parses. |
| 90 | + w.emitLocked() |
| 91 | + w.pendingCR = true |
| 92 | + default: |
| 93 | + w.buf = append(w.buf, b) |
| 94 | + if len(w.buf) >= maxPassthroughLineBytes { |
| 95 | + w.emitLocked() |
| 96 | + } |
| 97 | + } |
| 98 | + } |
| 99 | + return len(p), nil |
| 100 | +} |
| 101 | + |
| 102 | +func (w *passthroughWriter) emitLocked() { |
| 103 | + line := neutralizeWorkflowCommand(string(w.buf)) |
| 104 | + w.buf = w.buf[:0] |
| 105 | + _, _ = io.WriteString(w.framer.dst, PassthroughPrefix+line+"\n") |
| 106 | +} |
| 107 | + |
| 108 | +func (w *passthroughWriter) flushLocked() { |
| 109 | + if len(w.buf) > 0 { |
| 110 | + w.emitLocked() |
| 111 | + } |
| 112 | +} |
| 113 | + |
| 114 | +// neutralizeWorkflowCommand defuses a line that tries to open a GitHub Actions |
| 115 | +// workflow command (::error::, ::stop-commands::, ...). The frame prefix already |
| 116 | +// prevents the runner from seeing one, since a framed line no longer starts with |
| 117 | +// "::"; this is defense in depth for consumers that strip the prefix, and it |
| 118 | +// keeps the intent visible in the log rather than silently dropping it. |
| 119 | +func neutralizeWorkflowCommand(line string) string { |
| 120 | + trimmed := strings.TrimLeft(line, " \t") |
| 121 | + if !strings.HasPrefix(trimmed, "::") { |
| 122 | + return line |
| 123 | + } |
| 124 | + indent := line[:len(line)-len(trimmed)] |
| 125 | + return indent + "%3A%3A" + trimmed[2:] |
| 126 | +} |
0 commit comments