amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1//go:build linux
2
3package host
4
5import (
6 "context"
7 "crypto/rand"
8 "encoding/hex"
9 "encoding/json"
10 "errors"
11 "fmt"
12 "log"
13 "os"
14 "os/exec"
15 "path/filepath"
16 "strconv"
17 "strings"
18 "time"
19
20 "github.com/carlosframework/platform/internal/hostbox"
21 "github.com/carlosframework/platform/internal/safefile"
22 "golang.org/x/sys/unix"
23)
24
25var claimQuiesceOwner = claimStateOwner
26
27// claimEdgeMaskOwner records the identity of the runtime-mask object that the
28// migration put in place. It lives in root's own journal directory, which the
29// agent cannot write.
30var claimEdgeMaskOwner = "/var/lib/carlos-claims/edge-mask"
31
32var claimEdgeRuntimeMask = "/run/systemd/system/carlos-edge.service"
33
34type claimEdgeMaskRecord struct {
35 Version int `json:"v"`
36 Boot string `json:"boot"`
37 Token string `json:"token"`
38 Device uint64 `json:"device"`
39 Inode uint64 `json:"inode"`
40}
41
42var claimSystemctl = func(ctx context.Context, args ...string) ([]byte, error) {
43 return exec.CommandContext(ctx, "systemctl", args...).Output()
44}
45
46func claimEdgeMaskAnchor(token string) string {
47 return filepath.Join(filepath.Dir(claimEdgeRuntimeMask), ".carlos-edge.service.mask-"+token)
48}
49
50func currentClaimEdgeMask(path, boot, token string) (claimEdgeMaskRecord, error) {
51 dir, err := safefile.OpenDir(filepath.Dir(path))
52 if err != nil {
53 return claimEdgeMaskRecord{}, err
54 }
55 defer dir.Close()
56 return currentClaimEdgeMaskAt(dir, filepath.Base(path), boot, token)
57}
58
59func currentClaimEdgeMaskAt(dir *os.File, name, boot, token string) (claimEdgeMaskRecord, error) {
60 var st unix.Stat_t
61 if err := unix.Fstatat(int(dir.Fd()), name, &st, unix.AT_SYMLINK_NOFOLLOW); err != nil {
62 return claimEdgeMaskRecord{}, err
63 }
64 return claimEdgeMaskRecord{
65 Version: 1,
66 Boot: boot,
67 Token: token,
68 Device: uint64(st.Dev),
69 Inode: st.Ino,
70 }, nil
71}
72
73func decodeClaimEdgeMaskRecord(raw []byte, boot string) (claimEdgeMaskRecord, bool) {
74 var record claimEdgeMaskRecord
75 if json.Unmarshal(raw, &record) != nil {
76 return claimEdgeMaskRecord{}, false
77 }
78 token, tokenErr := hex.DecodeString(record.Token)
79 if tokenErr != nil || len(token) != 16 || record.Version != 1 || record.Boot != boot || record.Inode == 0 {
80 return claimEdgeMaskRecord{}, false
81 }
82 return record, true
83}
84
85func claimEdgeMaskStillOwned(record claimEdgeMaskRecord) (bool, error) {
86 dir, err := safefile.OpenDir(filepath.Dir(claimEdgeRuntimeMask))
87 if errors.Is(err, os.ErrNotExist) {
88 return false, nil
89 }
90 if err != nil {
91 return false, err
92 }
93 defer dir.Close()
94 anchor, err := currentClaimEdgeMaskAt(dir, filepath.Base(claimEdgeMaskAnchor(record.Token)), record.Boot, record.Token)
95 if errors.Is(err, os.ErrNotExist) {
96 return false, nil
97 }
98 if err != nil || anchor != record {
99 return false, err
100 }
101 current, err := currentClaimEdgeMaskAt(dir, filepath.Base(claimEdgeRuntimeMask), record.Boot, record.Token)
102 if errors.Is(err, os.ErrNotExist) {
103 return false, nil
104 }
105 return current == record, err
106}
107
108func writeClaimEdgeMaskRecord(record claimEdgeMaskRecord) error {
109 raw, err := json.Marshal(record)
110 if err != nil {
111 return err
112 }
113 return writeFileDurable(claimEdgeMaskOwner, append(raw, '\n'), 0600)
114}
115
116func discardClaimEdgeMaskRecord(record claimEdgeMaskRecord, recorded bool) error {
117 var anchorErr error
118 if recorded {
119 dir, err := safefile.OpenDir(filepath.Dir(claimEdgeRuntimeMask))
120 if err == nil {
121 defer dir.Close()
122 name := filepath.Base(claimEdgeMaskAnchor(record.Token))
123 var current claimEdgeMaskRecord
124 current, err = currentClaimEdgeMaskAt(dir, name, record.Boot, record.Token)
125 if err == nil && current == record {
126 if err = unix.Unlinkat(int(dir.Fd()), name, 0); err == nil {
127 err = dir.Sync()
128 }
129 }
130 }
131 if err != nil && !errors.Is(err, os.ErrNotExist) {
132 anchorErr = err
133 }
134 }
135 return errors.Join(anchorErr, removeClaimEdgeMaskRecord())
136}
137
138func removeClaimEdgeMaskRecord() error {
139 dir, err := safefile.OpenDir(filepath.Dir(claimEdgeMaskOwner))
140 if errors.Is(err, os.ErrNotExist) {
141 return nil
142 }
143 if err != nil {
144 return err
145 }
146 defer dir.Close()
147 err = unix.Unlinkat(int(dir.Fd()), filepath.Base(claimEdgeMaskOwner), 0)
148 if errors.Is(err, os.ErrNotExist) {
149 return nil
150 }
151 if err != nil {
152 return err
153 }
154 return dir.Sync()
155}
156
157func createClaimEdgeMask(ctx context.Context, boot string) (claimEdgeMaskRecord, error) {
158 tokenBytes := make([]byte, 16)
159 if _, err := rand.Read(tokenBytes); err != nil {
160 return claimEdgeMaskRecord{}, err
161 }
162 token := hex.EncodeToString(tokenBytes)
163 dir, err := safefile.OpenDir(filepath.Dir(claimEdgeRuntimeMask))
164 if err != nil {
165 return claimEdgeMaskRecord{}, err
166 }
167 defer dir.Close()
168 anchor := filepath.Base(claimEdgeMaskAnchor(token))
169 if err := unix.Symlinkat("/dev/null", int(dir.Fd()), anchor); err != nil {
170 return claimEdgeMaskRecord{}, err
171 }
172 record, err := currentClaimEdgeMaskAt(dir, anchor, boot, token)
173 if err != nil {
174 removeErr := unix.Unlinkat(int(dir.Fd()), anchor, 0)
175 if removeErr == nil {
176 removeErr = dir.Sync()
177 }
178 return claimEdgeMaskRecord{}, errors.Join(err, removeErr)
179 }
180 if err == nil {
181 err = writeClaimEdgeMaskRecord(record)
182 }
183 if err == nil {
184 err = unix.Linkat(int(dir.Fd()), anchor, int(dir.Fd()), filepath.Base(claimEdgeRuntimeMask), 0)
185 }
186 if err == nil {
187 err = dir.Sync()
188 }
189 if err == nil {
190 _, err = claimSystemctl(ctx, "daemon-reload")
191 }
192 if err == nil {
193 var ours bool
194 ours, err = claimEdgeMaskStillOwned(record)
195 if err == nil && !ours {
196 err = errors.New("claim journal migration: runtime mask identity changed")
197 }
198 }
199 if err != nil {
200 // If linking did not put our exact object at the unit name, cleanup
201 // removes only the private anchor and record. An object at the unit
202 // name is preserved.
203 ours, identityErr := claimEdgeMaskStillOwned(record)
204 if identityErr == nil && ours {
205 identityErr = releaseClaimEdgeMask(context.Background(), record)
206 } else {
207 identityErr = errors.Join(identityErr, discardClaimEdgeMaskRecord(record, record.Inode != 0))
208 }
209 return claimEdgeMaskRecord{}, errors.Join(err, identityErr)
210 }
211 return record, nil
212}
213
214// releaseClaimEdgeMask checks the object again immediately before asking
215// systemd to remove it. A missing or replaced object makes the ownership
216// record stale; it never grants authority over the object now at the path.
217func releaseClaimEdgeMask(ctx context.Context, record claimEdgeMaskRecord) error {
218 ours, err := claimEdgeMaskStillOwned(record)
219 if err != nil {
220 return err
221 }
222 if !ours {
223 return discardClaimEdgeMaskRecord(record, true)
224 }
225 // systemd has no compare-and-unlink operation. This check protects every
226 // replacement visible before the unmask call; a root operator concurrently
227 // replacing the name during this call is outside the migration protocol.
228 if _, err = claimSystemctl(ctx, "unmask", "--runtime", "carlos-edge.service"); err != nil {
229 return err
230 }
231 return discardClaimEdgeMaskRecord(record, true)
232}
233
234func claimEdgeStopped(ctx context.Context) error {
235 state, err := claimSystemctl(ctx, "show", "-p", "ActiveState", "--value", "carlos-edge.service")
236 if err != nil || (strings.TrimSpace(string(state)) != "inactive" && strings.TrimSpace(string(state)) != "failed") {
237 return errors.Join(err, hostbox.ErrTypedJournal)
238 }
239 return nil
240}
241
242// Migration is an offline upgrade. A running edge refuses before any mask is
243// written; otherwise runtime-mask the stopped edge, check again, and retain
244// the mask through the import. A live controller or same-UID child refuses
245// migration; the operator stops them before retrying.
246func quiesceLegacyClaimJournals(ctx context.Context) (func(), error) {
247 uid, _, err := claimQuiesceOwner()
248 if err != nil {
249 return nil, err
250 }
251 if err = claimEdgeStopped(ctx); err != nil {
252 return nil, err
253 }
254 boot, err := os.ReadFile("/proc/sys/kernel/random/boot_id")
255 if err != nil {
256 return nil, err
257 }
258 record, err := safefile.ReadFile(claimEdgeMaskOwner)
259 if err != nil && !errors.Is(err, os.ErrNotExist) {
260 return nil, err
261 }
262 before, err := claimSystemctl(ctx, "is-enabled", "carlos-edge.service")
263 state := strings.TrimSpace(string(before))
264 masked := state == "masked" || state == "masked-runtime"
265 if err != nil && !masked {
266 return nil, err
267 }
268 // Only the identity of the concrete runtime-mask object grants authority
269 // to lift it. A same-boot intent, malformed record, or replaced object is
270 // ambiguous and therefore leaves the mask in place.
271 owned, recorded := decodeClaimEdgeMaskRecord(record, strings.TrimSpace(string(boot)))
272 ours := false
273 if state == "masked-runtime" && recorded {
274 ours, err = claimEdgeMaskStillOwned(owned)
275 if err != nil {
276 return nil, err
277 }
278 }
279 if !ours && record != nil {
280 if err = discardClaimEdgeMaskRecord(owned, recorded); err != nil {
281 return nil, err
282 }
283 }
284 if !masked {
285 owned, err = createClaimEdgeMask(ctx, strings.TrimSpace(string(boot)))
286 if err != nil {
287 return nil, err
288 }
289 ours = true
290 }
291 release := func() {
292 if !ours {
293 return
294 }
295 if err := releaseClaimEdgeMask(context.Background(), owned); err != nil {
296 log.Printf("claim journal migration: edge runtime mask remains: %v", err)
297 }
298 }
299 // The edge may have started between the first check and the mask.
300 if err = claimEdgeStopped(ctx); err != nil {
301 release()
302 return nil, err
303 }
304 if err = scanQuiesceProcs("/proc", uid, os.Getpid()); err != nil {
305 release()
306 return nil, err
307 }
308 return release, nil
309}
310
311// scanQuiesceProcs refuses while any process runs as the service uid, except
312// this migration's own direct children: journal-migrate holds the registry
313// through a child it runs as the service user (openServiceRegistry), and
314// counting that child would make the migration refuse itself every time. The
315// refusal names the process it found.
316func scanQuiesceProcs(procDir string, uid, self int) error {
317 entries, err := os.ReadDir(procDir)
318 if err != nil {
319 return err
320 }
321 for _, entry := range entries {
322 pid, err := strconv.Atoi(entry.Name())
323 if err != nil {
324 continue
325 }
326 var st unix.Stat_t
327 if err = unix.Stat(filepath.Join(procDir, entry.Name()), &st); errors.Is(err, unix.ENOENT) {
328 continue
329 }
330 if err != nil {
331 return errors.Join(err, hostbox.ErrTypedJournal)
332 }
333 if int(st.Uid) != uid {
334 continue
335 }
336 ppid, perr := procParent(filepath.Join(procDir, entry.Name(), "stat"))
337 if errors.Is(perr, os.ErrNotExist) {
338 continue
339 }
340 if perr == nil && ppid == self {
341 continue
342 }
343 return errors.Join(errors.New("service-user process "+strconv.Itoa(pid)+" is still running"), perr, hostbox.ErrTypedJournal)
344 }
345 return nil
346}
347
348// procParent reads the parent pid from a /proc/<pid>/stat file. The command
349// name is parenthesised and may contain spaces, so the fields after the last
350// ')' are read: state, then ppid.
351func procParent(statPath string) (int, error) {
352 raw, err := os.ReadFile(statPath)
353 if err != nil {
354 return 0, err
355 }
356 ppid, err := procStatParent(raw)
357 if err != nil {
358 return 0, fmt.Errorf("%s: %w", statPath, err)
359 }
360 return ppid, nil
361}
362
363// procStatParent parses the ppid from a /proc stat line. The quiesce scan
364// and the generation census share it, so their own-child rule is one rule.
365func procStatParent(raw []byte) (int, error) {
366 s := string(raw)
367 i := strings.LastIndexByte(s, ')')
368 if i < 0 {
369 return 0, errors.New("malformed stat")
370 }
371 fields := strings.Fields(s[i+1:])
372 if len(fields) < 2 {
373 return 0, errors.New("malformed stat")
374 }
375 return strconv.Atoi(fields[1])
376}
377
378func recoverLegacyClaimEdgeMask(ctx context.Context) error {
379 record, err := safefile.ReadFile(claimEdgeMaskOwner)
380 if errors.Is(err, os.ErrNotExist) {
381 return nil
382 }
383 if err != nil {
384 return err
385 }
386 boot, err := os.ReadFile("/proc/sys/kernel/random/boot_id")
387 if err != nil {
388 return err
389 }
390 ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
391 defer cancel()
392 before, err := claimSystemctl(ctx, "is-enabled", "carlos-edge.service")
393 state := strings.TrimSpace(string(before))
394 if err != nil && state != "masked" && state != "masked-runtime" {
395 return err
396 }
397 owned, recorded := decodeClaimEdgeMaskRecord(record, strings.TrimSpace(string(boot)))
398 if state == "masked-runtime" && recorded {
399 return releaseClaimEdgeMask(ctx, owned)
400 }
401 return discardClaimEdgeMaskRecord(owned, recorded)
402}
403