| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "net" |
| 10 | "net/http" |
| 11 | "slices" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/controlsock" |
| 15 | "github.com/carlosframework/platform/internal/hostbox" |
| 16 | "github.com/carlosframework/platform/internal/hostclaim" |
| 17 | "github.com/carlosframework/platform/internal/hostguard" |
| 18 | "github.com/carlosframework/platform/internal/stopevidence" |
| 19 | ) |
| 20 | |
| 21 | // WritersStoppedRequest asks root for the writers-stopped observation of one |
| 22 | // exact writer session on this box (claim Plan 11, R3; L10). It names the |
| 23 | // session and the claim it was read at, and carries the agent's narrowing |
| 24 | // hint. It carries no pid, process start, path or cgroup: root takes the |
| 25 | // agent's process from the socket's peer credentials, the database paths |
| 26 | // from the installation inventory and the containment from the launch |
| 27 | // journal. |
| 28 | type WritersStoppedRequest struct { |
| 29 | Region string `json:"region"` |
| 30 | Host string `json:"host"` |
| 31 | Incarnation string `json:"incarnation"` |
| 32 | InstallationStamp uint64 `json:"installation_stamp"` |
| 33 | WriterSession uint64 `json:"writer_session"` |
| 34 | Box string `json:"box"` |
| 35 | Enrollment string `json:"enrollment"` |
| 36 | BootID string `json:"boot_id"` |
| 37 | ControllerID string `json:"controller_id"` |
| 38 | ExpectedClaimETag string `json:"expected_claim_etag"` |
| 39 | ExpectedDirectoryRevision string `json:"expected_directory_revision"` |
| 40 | RosterRevision uint64 `json:"roster_revision"` |
| 41 | Replicator []stopevidence.DescriptorHint `json:"replicator"` |
| 42 | } |
| 43 | |
| 44 | // WritersStoppedEvidence is root's answer: what its own observation found |
| 45 | // for the session. It is not stop evidence and is never fenced, signed or |
| 46 | // published; root's stop receipt still follows the release. |
| 47 | type WritersStoppedEvidence struct { |
| 48 | Host string `json:"host"` |
| 49 | Incarnation string `json:"incarnation"` |
| 50 | InstallationStamp uint64 `json:"installation_stamp"` |
| 51 | WriterSession uint64 `json:"writer_session"` |
| 52 | Box string `json:"box"` |
| 53 | Enrollment string `json:"enrollment"` |
| 54 | BootID string `json:"boot_id"` |
| 55 | ControllerID string `json:"controller_id"` |
| 56 | ClaimETag string `json:"claim_etag"` |
| 57 | ClaimGen uint64 `json:"claim_gen"` |
| 58 | AgentPID int `json:"agent_pid"` |
| 59 | AgentStart string `json:"agent_start"` |
| 60 | CheckedAt time.Time `json:"checked_at"` |
| 61 | CheckedPaths []string `json:"checked_paths"` |
| 62 | CgroupIdentities []string `json:"cgroup_identities"` |
| 63 | Retained []stopevidence.Descriptor `json:"retained"` |
| 64 | } |
| 65 | |
| 66 | // WritersKernelEvidence is the kernel half's observation. |
| 67 | type WritersKernelEvidence struct { |
| 68 | CheckedPaths []string |
| 69 | CgroupIdentities []string |
| 70 | Retained []stopevidence.Descriptor |
| 71 | } |
| 72 | |
| 73 | // stopWritersKernel is root's kernel half of the writers-stopped |
| 74 | // observation: the session's journaled spawns settled and its containment |
| 75 | // empty, and every holder of the installation's database files one of the |
| 76 | // hinted replicator descriptors in the authenticated agent process. |
| 77 | type stopWritersKernel interface { |
| 78 | VerifyWritersStopped(context.Context, hostguard.FenceTarget, StopInventory, StopPeer, []stopevidence.DescriptorHint) (WritersKernelEvidence, error) |
| 79 | } |
| 80 | |
| 81 | // WritersStopped is root's writers-stopped observation for req (R3). Root |
| 82 | // trusts no agent-authored list: the hint can only narrow what its own scan |
| 83 | // exempts, and the process it exempts in is the authenticated peer. |
| 84 | func (c ClaimStopCoordinator) WritersStopped(ctx context.Context, req WritersStoppedRequest) (WritersStoppedEvidence, error) { |
| 85 | if ctx == nil || c.CarlosUID == 0 || c.Peers == nil || c.Claims == nil || c.Inventory == nil || c.Kernel == nil || c.Now == nil || |
| 86 | !stopComponent(req.Host) || !stopComponent(req.Box) || !stopComponent(req.Enrollment) || req.Incarnation == "" || req.InstallationStamp == 0 || req.WriterSession == 0 || |
| 87 | req.BootID == "" || req.ControllerID == "" || req.ExpectedClaimETag == "" || req.ExpectedDirectoryRevision == "" || req.RosterRevision == 0 { |
| 88 | return WritersStoppedEvidence{}, ErrClaimStopUnavailable |
| 89 | } |
| 90 | kernel, ok := c.Kernel.(stopWritersKernel) |
| 91 | if !ok { |
| 92 | return WritersStoppedEvidence{}, ErrClaimStopUnavailable |
| 93 | } |
| 94 | peer, err := c.Peers.AuthenticatedPeer(ctx) |
| 95 | if err != nil { |
| 96 | return WritersStoppedEvidence{}, err |
| 97 | } |
| 98 | if peer.UID != c.CarlosUID || peer.PID <= 0 || peer.ProcessStart == "" || peer.BootID != req.BootID || peer.ControllerID != req.ControllerID { |
| 99 | return WritersStoppedEvidence{}, ErrClaimStopIdentity |
| 100 | } |
| 101 | observed, err := c.Claims.Read(ctx, req.Host) |
| 102 | if err != nil { |
| 103 | return WritersStoppedEvidence{}, err |
| 104 | } |
| 105 | if observed.Absent || observed.Claim == nil || observed.Claim.Holder == nil || observed.ETag != req.ExpectedClaimETag { |
| 106 | return WritersStoppedEvidence{}, ErrClaimStopIdentity |
| 107 | } |
| 108 | claim, holder := observed.Claim, observed.Claim.Holder |
| 109 | target := hostguard.FenceTarget{Region: req.Region, ClaimGen: claim.Gen, Host: req.Host, Incarnation: req.Incarnation, Stamp: req.InstallationStamp, Session: req.WriterSession, |
| 110 | Op: claim.Op.ID, Phase: claim.Op.Phase, Box: req.Box, Enrollment: req.Enrollment, BootID: req.BootID, ControllerID: req.ControllerID, ClaimETag: req.ExpectedClaimETag, DirectoryRevision: req.ExpectedDirectoryRevision, RosterRevision: req.RosterRevision} |
| 111 | // The session is this box's writer: the holder names it, and the claim |
| 112 | // gives this box the authority a stop of it would need. |
| 113 | if claim.Host != req.Host || holder.WriterBox != req.Box || !hostguard.StopAuthorityMatches(claim, target) { |
| 114 | return WritersStoppedEvidence{}, ErrClaimStopIdentity |
| 115 | } |
| 116 | inv, err := c.Inventory.ReadStopInventory(ctx, req.Host) |
| 117 | if err != nil { |
| 118 | return WritersStoppedEvidence{}, err |
| 119 | } |
| 120 | if inv.Host != req.Host || inv.Incarnation != req.Incarnation || inv.InstallationStamp != req.InstallationStamp || inv.Box != req.Box || inv.Enrollment != req.Enrollment || inv.DirectoryRevision != req.ExpectedDirectoryRevision { |
| 121 | return WritersStoppedEvidence{}, ErrClaimStopInventory |
| 122 | } |
| 123 | seen, err := kernel.VerifyWritersStopped(ctx, target, inv, peer, slices.Clone(req.Replicator)) |
| 124 | if err != nil { |
| 125 | return WritersStoppedEvidence{}, err |
| 126 | } |
| 127 | // The observation holds for the claim it was asked at. |
| 128 | again, err := c.Claims.Read(ctx, req.Host) |
| 129 | if err != nil { |
| 130 | return WritersStoppedEvidence{}, err |
| 131 | } |
| 132 | if again.ETag != req.ExpectedClaimETag { |
| 133 | return WritersStoppedEvidence{}, ErrClaimStopIdentity |
| 134 | } |
| 135 | return WritersStoppedEvidence{Host: req.Host, Incarnation: req.Incarnation, InstallationStamp: req.InstallationStamp, WriterSession: req.WriterSession, Box: req.Box, Enrollment: req.Enrollment, |
| 136 | BootID: req.BootID, ControllerID: req.ControllerID, ClaimETag: req.ExpectedClaimETag, ClaimGen: claim.Gen, AgentPID: peer.PID, AgentStart: peer.ProcessStart, |
| 137 | CheckedAt: c.Now().UTC(), CheckedPaths: seen.CheckedPaths, CgroupIdentities: seen.CgroupIdentities, Retained: seen.Retained}, nil |
| 138 | } |
| 139 | |
| 140 | // serveWritersStopped answers POST /claim/writers-stopped for an |
| 141 | // authenticated peer. |
| 142 | func (h ClaimStopHTTP) serveWritersStopped(w http.ResponseWriter, r *http.Request, peer StopPeer) { |
| 143 | decoder := json.NewDecoder(io.LimitReader(r.Body, 64<<10)) |
| 144 | decoder.DisallowUnknownFields() |
| 145 | var req WritersStoppedRequest |
| 146 | if decoder.Decode(&req) != nil || decoder.Decode(new(any)) != io.EOF { |
| 147 | http.Error(w, "refused", http.StatusBadRequest) |
| 148 | return |
| 149 | } |
| 150 | c := h.Coordinator |
| 151 | c.Peers = stopPeerValue{peer} |
| 152 | ev, err := c.WritersStopped(r.Context(), req) |
| 153 | if err != nil { |
| 154 | http.Error(w, "refused", http.StatusConflict) |
| 155 | return |
| 156 | } |
| 157 | w.Header().Set("Content-Type", "application/json") |
| 158 | _ = json.NewEncoder(w).Encode(ev) |
| 159 | } |
| 160 | |
| 161 | // CollectWritersStopped asks root over the instance-control socket at path |
| 162 | // for the writers-stopped observation of req's session. |
| 163 | func CollectWritersStopped(ctx context.Context, path string, req WritersStoppedRequest) (WritersStoppedEvidence, error) { |
| 164 | ctx, cancel := context.WithTimeout(ctx, 2*time.Minute) |
| 165 | defer cancel() |
| 166 | body, err := json.Marshal(req) |
| 167 | if err != nil { |
| 168 | return WritersStoppedEvidence{}, err |
| 169 | } |
| 170 | transport := &http.Transport{DisableKeepAlives: true, DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { |
| 171 | return (&net.Dialer{}).DialContext(ctx, "unix", path) |
| 172 | }} |
| 173 | defer transport.CloseIdleConnections() |
| 174 | request, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://control/claim/writers-stopped", bytes.NewReader(body)) |
| 175 | if err != nil { |
| 176 | return WritersStoppedEvidence{}, err |
| 177 | } |
| 178 | response, err := (&http.Client{Transport: transport}).Do(request) |
| 179 | if err != nil { |
| 180 | return WritersStoppedEvidence{}, err |
| 181 | } |
| 182 | defer response.Body.Close() |
| 183 | if response.StatusCode != http.StatusOK { |
| 184 | return WritersStoppedEvidence{}, fmt.Errorf("%w: writers-stopped: %s", ErrClaimStopUnavailable, response.Status) |
| 185 | } |
| 186 | decoder := json.NewDecoder(io.LimitReader(response.Body, 1<<20)) |
| 187 | decoder.DisallowUnknownFields() |
| 188 | var ev WritersStoppedEvidence |
| 189 | if err := decoder.Decode(&ev); err != nil { |
| 190 | return WritersStoppedEvidence{}, err |
| 191 | } |
| 192 | if ev.Host != req.Host || ev.Incarnation != req.Incarnation || ev.InstallationStamp != req.InstallationStamp || ev.WriterSession != req.WriterSession || ev.Box != req.Box || |
| 193 | ev.Enrollment != req.Enrollment || ev.BootID != req.BootID || ev.ControllerID != req.ControllerID || ev.ClaimETag != req.ExpectedClaimETag || ev.CheckedAt.IsZero() { |
| 194 | return WritersStoppedEvidence{}, fmt.Errorf("%w: writers-stopped evidence names another session", ErrClaimStopIdentity) |
| 195 | } |
| 196 | return ev, nil |
| 197 | } |
| 198 | |
| 199 | // WritersStopped is the agent's whole writers-stopped step for host's |
| 200 | // admitted attempt: StopWriters (idempotent), then root's observation of |
| 201 | // the witnessed session at the current claim, with the replicator hint |
| 202 | // recorded when it opened. It never releases the claim or the lease and |
| 203 | // publishes nothing. |
| 204 | func (a ClaimAssembly) WritersStopped(ctx context.Context, host string) (ClaimWritersDrained, WritersStoppedEvidence, error) { |
| 205 | read := func(ctx context.Context, host string) (hostclaim.Observed, error) { |
| 206 | return hostclaim.Read(ctx, a.Claims, host) |
| 207 | } |
| 208 | return a.writersStopped(ctx, host, read, func(ctx context.Context, req WritersStoppedRequest) (WritersStoppedEvidence, error) { |
| 209 | return CollectWritersStopped(ctx, controlsock.Path, req) |
| 210 | }) |
| 211 | } |
| 212 | |
| 213 | func (a ClaimAssembly) writersStopped(ctx context.Context, host string, read func(context.Context, string) (hostclaim.Observed, error), collect func(context.Context, WritersStoppedRequest) (WritersStoppedEvidence, error)) (ClaimWritersDrained, WritersStoppedEvidence, error) { |
| 214 | if a.Wake == nil || read == nil || collect == nil { |
| 215 | return ClaimWritersDrained{}, WritersStoppedEvidence{}, ErrClaimStopUnavailable |
| 216 | } |
| 217 | att := a.Wake.get(host) |
| 218 | if att == nil { |
| 219 | return ClaimWritersDrained{}, WritersStoppedEvidence{}, ErrClaimStopUnavailable |
| 220 | } |
| 221 | drained, err := att.StopWriters(ctx) |
| 222 | if err != nil { |
| 223 | return ClaimWritersDrained{}, WritersStoppedEvidence{}, err |
| 224 | } |
| 225 | observed, err := read(ctx, host) |
| 226 | if err != nil { |
| 227 | return ClaimWritersDrained{}, WritersStoppedEvidence{}, err |
| 228 | } |
| 229 | if err := drained.For(observed); err != nil { |
| 230 | return ClaimWritersDrained{}, WritersStoppedEvidence{}, err |
| 231 | } |
| 232 | local, err := a.localStopIdentity(ctx) |
| 233 | if err != nil { |
| 234 | return ClaimWritersDrained{}, WritersStoppedEvidence{}, err |
| 235 | } |
| 236 | t := drained.Target |
| 237 | if t.Box != local.Actor.Box || t.Enrollment != local.Actor.Enrollment || t.BootID != local.BootID || t.ControllerID != local.ControllerID { |
| 238 | return ClaimWritersDrained{}, WritersStoppedEvidence{}, hostbox.ErrChanged |
| 239 | } |
| 240 | revision := t.DirectoryRevision |
| 241 | if revision == "" { |
| 242 | revision = local.Actor.Binding |
| 243 | } |
| 244 | rosterRevision := t.RosterRevision |
| 245 | if rosterRevision == 0 { |
| 246 | rosterRevision = local.Actor.Revision |
| 247 | } |
| 248 | att.mu.Lock() |
| 249 | hints := slices.Clone(att.replicatorFDs) |
| 250 | att.mu.Unlock() |
| 251 | req := WritersStoppedRequest{Region: local.Actor.Region, Host: host, Incarnation: t.Incarnation, InstallationStamp: t.Stamp, WriterSession: t.Session, Box: t.Box, Enrollment: t.Enrollment, |
| 252 | BootID: t.BootID, ControllerID: t.ControllerID, ExpectedClaimETag: observed.ETag, ExpectedDirectoryRevision: revision, RosterRevision: rosterRevision, Replicator: hints} |
| 253 | ev, err := collect(ctx, req) |
| 254 | if err != nil { |
| 255 | return ClaimWritersDrained{}, WritersStoppedEvidence{}, err |
| 256 | } |
| 257 | return drained, ev, nil |
| 258 | } |
| 259 | |
amadan