carlos / platform Public

Download this file

1	//go:build linux
2
3	// Package maintenanceexec supplies process and pipe mechanics. Its result is
4	// not an approval, database-compatibility receipt or permission to resume.
5	package maintenanceexec
6
7	import (
8	"context"
9	"crypto/sha256"
10	"encoding/hex"
11	"io"
12	"os"
13	"os/exec"
14	"time"
15
16	"github.com/carlosframework/platform/internal/cgroup"
17	"github.com/carlosframework/platform/internal/lease"
18	"github.com/carlosframework/platform/internal/maintenance"
19	"golang.org/x/sys/unix"
20	)
21
22	const stopMargin = 2 * time.Second
23
24	type containment interface {
25	Start(*exec.Cmd) error
26	Populated() (bool, error)
27	KillAndEmpty(context.Context) error
28	}
29	type ownership interface {
30	SafeUntil() (time.Time, error)
31	Lost() <-chan struct{}
32	}
33
34	// Run requires concrete retained containment and lease handles. The caller
35	// must already hold the lifecycle fence, checked approval and verified release
36	// and environment snapshots. This package exposes no HTTP or CLI entry point.
37	// Artifact, containment and lease ownership remain with the caller on return.
38	func Run(ctx context.Context, group *cgroup.Prepared, held *lease.Held, artifact *os.File, launch maintenance.LaunchContext, input maintenance.Input, environment []string) ([]byte, error) {
39	if group == nil || held == nil {
40	return nil, ErrRefused
41	}
42	return run(ctx, group, held, artifact, launch, input, environment)
43	}
44
45	func run(ctx context.Context, group containment, held ownership, artifact *os.File, launch maintenance.LaunchContext, input maintenance.Input, environment []string) (result []byte, err error) {
46	if ctx == nil || ctx.Err() != nil || artifact == nil || launch.Version != 1 || len(input.Arguments) > maintenance.MaxArguments || len(input.Stdin) > maintenance.MaxInputBytes {
47	return nil, ErrRefused
48	}
49	input.Arguments = append([]string(nil), input.Arguments...)
50	input.Stdin = append([]byte(nil), input.Stdin...)
51	defer clear(input.Stdin)
52	raw, e := maintenance.EncodeLaunch(launch)
53	if e != nil {
54	return nil, ErrRefused
55	}
56	defer clear(raw)
57	frame, e := maintenance.EncodeInput(input)
58	if e != nil {
59	return nil, ErrRefused
60	}
61	clear(frame)
62	sum := sha256.Sum256(input.Stdin)
63	if hex.EncodeToString(sum[:]) != launch.InputSHA256 {
64	return nil, ErrRefused
65	}
66	executionDeadline, _ := time.Parse(time.RFC3339Nano, launch.ExecutionDeadline)
67	until, e := held.SafeUntil()
68	if e != nil {
69	return nil, ErrRefused
70	}
71	if !time.Now().Add(stopMargin).Before(until) || !time.Now().Add(stopMargin).Before(executionDeadline) {
72	return nil, ErrRefused
73	}
74	hashDeadline := until.Add(-stopMargin)
75	if executionDeadline.Add(-stopMargin).Before(hashDeadline) {
76	hashDeadline = executionDeadline.Add(-stopMargin)
77	}
78	hashCtx, cancelHash := context.WithDeadline(ctx, hashDeadline)
79	e = checkArtifact(hashCtx, artifact, launch.ArtifactDigest)
80	cancelHash()
81	if e != nil {
82	return nil, ErrRefused
83	}
84	pipes, e := newPipes()
85	if e != nil {
86	return nil, ErrRefused
87	}
88	defer pipes.close()
89	null, e := os.OpenFile(os.DevNull, os.O_WRONLY, 0)
90	if e != nil {
91	return nil, ErrRefused
92	}
93	defer null.Close()
94	cmd := exec.Command("/proc/self/fd/4", input.Arguments...)
95	cmd.Env = append([]string{}, environment...)
96	cmd.Dir = "/"
97	cmd.Stdin = pipes.inR
98	cmd.Stdout = pipes.outW
99	cmd.Stderr = null
100	cmd.ExtraFiles = []*os.File{pipes.contextR, artifact}
101	if until, e = held.SafeUntil(); e != nil || ctx.Err() != nil || !time.Now().Add(stopMargin).Before(until) || !time.Now().Add(stopMargin).Before(executionDeadline) {
102	return nil, ErrRefused
103	}
104	if e = group.Start(cmd); e != nil {
105	stopErr := stop(group, until)
106	if stopErr != nil {
107	return nil, ErrUncertain
108	}
109	return nil, ErrRefused
110	}
111	pipes.inR.Close()
112	pipes.outW.Close()
113	pipes.contextR.Close()
114	cleanupDone := false
115	defer func() {
116	if cleanupDone {
117	return
118	}
119	pipes.close()
120	if stop(group, until) != nil {
121	err = ErrUncertain
122	result = nil
123	}
124	}()
125	events := make(chan pipeEvent, 4)
126	go writePipe(pipes.inW, append([]byte(nil), input.Stdin...), "input", events)
127	go writePipe(pipes.contextW, append([]byte(nil), raw...), "context", events)
128	go func() {
129	b, e := io.ReadAll(io.LimitReader(pipes.outR, maintenance.MaxResultBytes+1))
130	if len(b) > maintenance.MaxResultBytes {
131	clear(b)
132	b = nil
133	e = ErrRefused
134	}
135	events <- pipeEvent{"output", b, e}
136	}()
137	go func() { events <- pipeEvent{"wait", nil, cmd.Wait()} }()
138	tick := time.NewTicker(10 * time.Millisecond)
139	defer tick.Stop()
140	left := 4
141	failed := false
142	waited := false
143	for left > 0 {
144	select {
145	case event := <-events:
146	left--
147	if event.kind == "wait" {
148	waited = true
149	}
150	if event.err != nil {
151	failed = true
152	}
153	if event.kind == "output" {
154	result = event.data
155	}
156	case <-ctx.Done():
157	failed = true
158	case <-held.Lost():
159	failed = true
160	case <-tick.C:
161	next, e := held.SafeUntil()
162	if e != nil {
163	failed = true
164	} else {
165	until = next
166	}
167	now := time.Now()
168	if !now.Add(stopMargin).Before(until) || !now.Add(stopMargin).Before(executionDeadline) {
169	failed = true
170	}
171	}
172	if failed || waited {
173	if waited {
174	populated, e := group.Populated()
175	if e != nil || populated {
176	failed = true
177	}
178	}
179	e := stop(group, until)
180	if e != nil {
181	clear(result)
182	return nil, ErrUncertain
183	}
184	cleanupDone = true
185	// Closing parent pipe endpoints also releases blocked I/O after the
186	// kernel has proved the child tree empty.
187	if failed {
188	pipes.close()
189	}
190	for left > 0 {
191	select {
192	case event := <-events:
193	left--
194	if event.err != nil {
195	failed = true
196	}
197	if event.kind == "output" {
198	result = event.data
199	}
200	case <-time.After(time.Until(until)):
201	clear(result)
202	return nil, ErrUncertain
203	}
204	}
205	break
206	}
207	}
208	if !cleanupDone {
209	e := stop(group, until)
210	if e != nil {
211	clear(result)
212	return nil, ErrUncertain
213	}
214	cleanupDone = true
215	}
216	if _, e = held.SafeUntil(); e != nil || ctx.Err() != nil || !time.Now().Before(executionDeadline) {
217	failed = true
218	}
219	if failed || len(result) == 0 {
220	clear(result)
221	return nil, ErrRefused
222	}
223	return result, nil
224	}
225
226	func stop(group containment, until time.Time) error {
227	deadline := time.Now().Add(stopMargin)
228	if until.After(deadline) {
229	deadline = until
230	}
231	ctx, cancel := context.WithDeadline(context.Background(), deadline)
232	defer cancel()
233	err := group.KillAndEmpty(ctx)
234	if err != nil || !time.Now().Before(until) {
235	return ErrUncertain
236	}
237	return nil
238	}
239
240	type pipeEvent struct {
241	kind string
242	data []byte
243	err error
244	}
245
246	func writePipe(f *os.File, data []byte, kind string, events chan<- pipeEvent) {
247	defer clear(data)
248	_, err := f.Write(data)
249	closeErr := f.Close()
250	if err == nil {
251	err = closeErr
252	}
253	events <- pipeEvent{kind: kind, err: err}
254	}
255	func checkArtifact(ctx context.Context, f *os.File, want string) error {
256	flags, e := unix.FcntlInt(f.Fd(), unix.F_GETFL, 0)
257	if e != nil || flags&unix.O_ACCMODE != unix.O_RDONLY {
258	return ErrRefused
259	}
260	st, e := f.Stat()
261	if e != nil || !st.Mode().IsRegular() || st.Size() <= 0 {
262	return ErrRefused
263	}
264	hash := sha256.New()
265	reader := io.NewSectionReader(f, 0, st.Size())
266	buf := make([]byte, 64<<10)
267	for {
268	if ctx.Err() != nil {
269	return ErrRefused
270	}
271	n, e := reader.Read(buf)
272	if n > 0 {
273	hash.Write(buf[:n])
274	}
275	if e == io.EOF {
276	break
277	}
278	if e != nil {
279	return ErrRefused
280	}
281	}
282	after, e := f.Stat()
283	if e != nil || after.Size() != st.Size() || !os.SameFile(st, after) || hex.EncodeToString(hash.Sum(nil)) != want {
284	return ErrRefused
285	}
286	return nil
287	}
288