| 1 | package registry |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/gob" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "os" |
| 11 | "sync" |
| 12 | "time" |
| 13 | ) |
| 14 | |
| 15 | var errRemoteMethod = errors.New("registry: unsupported remote operation") |
| 16 | |
| 17 | var remoteErrors = []error{os.ErrNotExist, os.ErrPermission, sql.ErrNoRows, ErrClaimOwnership, ErrTimerChanged, ErrTimerKind, ErrPathsChanged, ErrFleetBoxActive, context.Canceled, context.DeadlineExceeded, ErrGoodBuildChanged} |
| 18 | |
| 19 | func responseError(out remoteResponse) error { |
| 20 | if !out.Failed { |
| 21 | return nil |
| 22 | } |
| 23 | if out.ErrorKind > 0 && out.ErrorKind <= len(remoteErrors) { |
| 24 | sentinel := remoteErrors[out.ErrorKind-1] |
| 25 | if out.Error == sentinel.Error() { |
| 26 | return sentinel |
| 27 | } |
| 28 | return fmt.Errorf("%s: %w", out.Error, sentinel) |
| 29 | } |
| 30 | return errors.New(out.Error) |
| 31 | } |
| 32 | |
| 33 | func setResponseError(out *remoteResponse, err error) { |
| 34 | if err == nil { |
| 35 | return |
| 36 | } |
| 37 | out.Failed = true |
| 38 | out.Error = err.Error() |
| 39 | for i, sentinel := range remoteErrors { |
| 40 | if errors.Is(err, sentinel) { |
| 41 | out.ErrorKind = i + 1 |
| 42 | break |
| 43 | } |
| 44 | } |
| 45 | } |
| 46 | |
| 47 | type remoteClient struct { |
| 48 | mu sync.Mutex |
| 49 | enc *gob.Encoder |
| 50 | dec *gob.Decoder |
| 51 | input io.WriteCloser |
| 52 | output io.ReadCloser |
| 53 | wait func() error |
| 54 | closed bool |
| 55 | connect func() (io.WriteCloser, io.ReadCloser, func() error, error) |
| 56 | } |
| 57 | |
| 58 | // OpenRemote connects to an already privilege-dropped registry worker. |
| 59 | func OpenRemote(path string, input io.WriteCloser, output io.ReadCloser, wait func() error) (*Registry, error) { |
| 60 | c := &remoteClient{} |
| 61 | if err := c.attach(input, output, wait); err != nil { |
| 62 | return nil, err |
| 63 | } |
| 64 | return &Registry{path: path, remote: c}, nil |
| 65 | } |
| 66 | |
| 67 | // OpenRemoteReconnect reads service-controlled answers, not an authority boundary. |
| 68 | // Failed or cancelled mutations are never retried; the next call reconnects. |
| 69 | // Canceled does not mean not applied: cancellation can race a committed mutation. |
| 70 | func OpenRemoteReconnect(path string, connect func() (io.WriteCloser, io.ReadCloser, func() error, error)) (*Registry, error) { |
| 71 | c := &remoteClient{connect: connect} |
| 72 | if err := c.reconnect(context.Background()); err != nil { |
| 73 | return nil, err |
| 74 | } |
| 75 | return &Registry{path: path, remote: c}, nil |
| 76 | } |
| 77 | |
| 78 | func (c *remoteClient) reconnect(ctx context.Context) (err error) { |
| 79 | input, output, wait, err := c.connect() |
| 80 | if err != nil { |
| 81 | return err |
| 82 | } |
| 83 | finished := make(chan struct{}) |
| 84 | stop := context.AfterFunc(ctx, func() { input.Close(); output.Close(); close(finished) }) |
| 85 | defer func() { |
| 86 | if !stop() { |
| 87 | <-finished |
| 88 | } |
| 89 | if ctx.Err() != nil { |
| 90 | c.disconnect() |
| 91 | err = ctx.Err() |
| 92 | } |
| 93 | }() |
| 94 | return c.attach(input, output, wait) |
| 95 | } |
| 96 | |
| 97 | func (c *remoteClient) attach(input io.WriteCloser, output io.ReadCloser, wait func() error) error { |
| 98 | c.input, c.output, c.wait = input, output, wait |
| 99 | c.enc, c.dec = gob.NewEncoder(input), gob.NewDecoder(output) |
| 100 | var ready remoteResponse |
| 101 | err := c.dec.Decode(&ready) |
| 102 | if err == nil { |
| 103 | err = responseError(ready) |
| 104 | } |
| 105 | if err != nil { |
| 106 | c.disconnect() |
| 107 | } |
| 108 | return err |
| 109 | } |
| 110 | |
| 111 | func (c *remoteClient) disconnect() error { |
| 112 | if c.input == nil { |
| 113 | return nil |
| 114 | } |
| 115 | c.input.Close() |
| 116 | c.output.Close() |
| 117 | c.input, c.output = nil, nil |
| 118 | if c.wait != nil { |
| 119 | return c.wait() |
| 120 | } |
| 121 | return nil |
| 122 | } |
| 123 | |
| 124 | func (c *remoteClient) stop() error { |
| 125 | if c.closed { |
| 126 | return nil |
| 127 | } |
| 128 | c.closed = true |
| 129 | return c.disconnect() |
| 130 | } |
| 131 | |
| 132 | func (c *remoteClient) close() error { |
| 133 | c.mu.Lock() |
| 134 | defer c.mu.Unlock() |
| 135 | return c.stop() |
| 136 | } |
| 137 | |
| 138 | func (c *remoteClient) call(ctx context.Context, q remoteRequest, callback func(remoteResponse) error) (out remoteResponse, err error) { |
| 139 | c.mu.Lock() |
| 140 | defer c.mu.Unlock() |
| 141 | if c.closed { |
| 142 | return out, os.ErrClosed |
| 143 | } |
| 144 | if ctx != nil { |
| 145 | if err = ctx.Err(); err != nil { |
| 146 | return |
| 147 | } |
| 148 | if deadline, ok := ctx.Deadline(); ok { |
| 149 | q.Deadline = deadline.UnixNano() |
| 150 | } |
| 151 | } |
| 152 | if c.input == nil { |
| 153 | if c.connect == nil { |
| 154 | return out, os.ErrClosed |
| 155 | } |
| 156 | reconnectContext := ctx |
| 157 | if reconnectContext == nil { |
| 158 | reconnectContext = context.Background() |
| 159 | } |
| 160 | if err = c.reconnect(reconnectContext); err != nil { |
| 161 | return |
| 162 | } |
| 163 | } |
| 164 | // Closing the pipe cancels the worker context, including an in-flight SQL call. |
| 165 | if ctx != nil { |
| 166 | input, output := c.input, c.output |
| 167 | stopped := make(chan struct{}) |
| 168 | stop := context.AfterFunc(ctx, func() { input.Close(); output.Close(); close(stopped) }) |
| 169 | defer func() { |
| 170 | if !stop() { |
| 171 | <-stopped |
| 172 | } |
| 173 | if ctx.Err() != nil { |
| 174 | err = ctx.Err() |
| 175 | c.disconnect() |
| 176 | } |
| 177 | }() |
| 178 | } |
| 179 | // A callback panic or a broken pipe must release any child transaction. |
| 180 | complete := false |
| 181 | defer func() { |
| 182 | if !complete { |
| 183 | c.disconnect() |
| 184 | } |
| 185 | }() |
| 186 | if err = c.enc.Encode(q); err != nil { |
| 187 | return |
| 188 | } |
| 189 | if err = c.dec.Decode(&out); err != nil { |
| 190 | return |
| 191 | } |
| 192 | var callbackErr error |
| 193 | if out.Callback != "" { |
| 194 | if callback == nil { |
| 195 | return out, errRemoteMethod |
| 196 | } |
| 197 | callbackErr = callback(out) |
| 198 | if ctx != nil && ctx.Err() != nil { |
| 199 | return out, ctx.Err() |
| 200 | } |
| 201 | ack := remoteRequest{Method: "callback"} |
| 202 | if callbackErr != nil { |
| 203 | ack.CallbackFailed = true |
| 204 | ack.CallbackError = callbackErr.Error() |
| 205 | } |
| 206 | if err = c.enc.Encode(ack); err != nil { |
| 207 | return |
| 208 | } |
| 209 | out = remoteResponse{} |
| 210 | if err = c.dec.Decode(&out); err != nil { |
| 211 | return |
| 212 | } |
| 213 | if out.Callback != "" { |
| 214 | return out, errRemoteMethod |
| 215 | } |
| 216 | } |
| 217 | complete = true |
| 218 | if callbackErr != nil { |
| 219 | return out, callbackErr |
| 220 | } |
| 221 | return out, responseError(out) |
| 222 | } |
| 223 | |
| 224 | // ServeRemote owns SQLite in the worker. Callers must drop credentials first. |
| 225 | func ServeRemote(path string, input io.Reader, output io.Writer) error { |
| 226 | return serveRemote(path, input, output, false) |
| 227 | } |
| 228 | |
| 229 | func ServeRemoteReadOnly(path string, input io.Reader, output io.Writer) error { |
| 230 | return serveRemote(path, input, output, true) |
| 231 | } |
| 232 | |
| 233 | func serveRemote(path string, input io.Reader, output io.Writer, readOnly bool) error { |
| 234 | if os.Geteuid() == 0 { |
| 235 | return errors.New("registry: remote worker refuses root") |
| 236 | } |
| 237 | enc, dec := gob.NewEncoder(output), gob.NewDecoder(input) |
| 238 | var r *Registry |
| 239 | var err error |
| 240 | if readOnly { |
| 241 | r, err = OpenReadOnly(path) |
| 242 | } else { |
| 243 | r, err = Open(path) |
| 244 | } |
| 245 | ready := remoteResponse{} |
| 246 | setResponseError(&ready, err) |
| 247 | if writeErr := enc.Encode(ready); writeErr != nil { |
| 248 | if r != nil { |
| 249 | r.Close() |
| 250 | } |
| 251 | return writeErr |
| 252 | } |
| 253 | if err != nil { |
| 254 | return err |
| 255 | } |
| 256 | defer r.Close() |
| 257 | workerCtx, stopWorker := context.WithCancel(context.Background()) |
| 258 | defer stopWorker() |
| 259 | requests := make(chan remoteRequest) |
| 260 | readErrors := make(chan error, 1) |
| 261 | go func() { |
| 262 | defer close(requests) |
| 263 | defer stopWorker() |
| 264 | for { |
| 265 | var q remoteRequest |
| 266 | if err := dec.Decode(&q); err != nil { |
| 267 | readErrors <- err |
| 268 | return |
| 269 | } |
| 270 | select { |
| 271 | case requests <- q: |
| 272 | case <-workerCtx.Done(): |
| 273 | return |
| 274 | } |
| 275 | } |
| 276 | }() |
| 277 | for q := range requests { |
| 278 | if readOnly && q.Method != "Get" && q.Method != "TimersFor" && q.Method != "ArtifactHosts" { |
| 279 | out := remoteResponse{} |
| 280 | setResponseError(&out, errRemoteMethod) |
| 281 | if err := enc.Encode(out); err != nil { |
| 282 | return err |
| 283 | } |
| 284 | continue |
| 285 | } |
| 286 | ctx := workerCtx |
| 287 | cancel := func() {} |
| 288 | if q.Deadline != 0 { |
| 289 | ctx, cancel = context.WithDeadline(ctx, time.Unix(0, q.Deadline)) |
| 290 | } |
| 291 | out, callErr := r.dispatchRemote(ctx, q, func(event remoteResponse) error { |
| 292 | if err := enc.Encode(event); err != nil { |
| 293 | return err |
| 294 | } |
| 295 | ack, ok := <-requests |
| 296 | if !ok { |
| 297 | return workerCtx.Err() |
| 298 | } |
| 299 | if ack.Method != "callback" { |
| 300 | return errRemoteMethod |
| 301 | } |
| 302 | if ack.CallbackFailed { |
| 303 | return errors.New(ack.CallbackError) |
| 304 | } |
| 305 | return nil |
| 306 | }) |
| 307 | cancel() |
| 308 | setResponseError(&out, callErr) |
| 309 | if err = enc.Encode(out); err != nil { |
| 310 | return err |
| 311 | } |
| 312 | } |
| 313 | select { |
| 314 | case err := <-readErrors: |
| 315 | if errors.Is(err, io.EOF) { |
| 316 | return nil |
| 317 | } |
| 318 | return err |
| 319 | default: |
| 320 | return workerCtx.Err() |
| 321 | } |
| 322 | |
| 323 | } |
| 324 | |
amadan