openvibers/openvibe.link Commits
Refuse jobs with inputs and enforce the declared net deny default
Changes
@@@ -237,7 +237,7 @@ Otherwise it advertises no class and refuses every well-formed job with `class n| direction | type | fields ||------------------|----------------|-------------------------------------------------------------------------------------------|--| server → device | `job` | `job` {`id` (`job_<ULID>`: the idempotency and ack key), `class`, `artifact` {`name`, `version`} (required for `function` and `code`, one exact version), `args` {}, `ttl_ms`, `limits` {`wall_ms`, `cpu_ms`, `mem_bytes`}, `net` (`deny`: no network whatever the Node's `worker.egress`; absent: `worker.egress` applies, `none` by default)} |++| server → device | `job` | `job` {`id` (`job_<ULID>`: the idempotency and ack key), `class`, `artifact` {`name`, `version`} (required for `function` and `code`, one exact version), `args` {}, `ttl_ms`, `limits` {`wall_ms`, `cpu_ms`, `mem_bytes`}, `net` (`deny`, `none`, `public`, `openvibe-only`; absent means `deny`, which the Node never widens to its own `worker.egress`; `public` and `openvibe-only` run under the stricter of what is asked and what `worker.egress` allows, and are refused when it is `none`), `inputs` (files pinned by Media id and sha256: **refused**, `job inputs not supported`, until the Node can stage and verify them)} || server → device | `job_cancel` | `id` || server → device | `job_exit_ack` | `id` || device → server | `job_started` | `id`, `started_ms` |@@@ -254,7 +254,9 @@ checked in this order:| `malformed job` | `bad_frame` | the `job` object does not decode (a field of the wrong type) || `unknown class` | `unsupported` | `class` is not one of `function`, `code`, `browser`, `linux`, `desktop`, `gpu` || `function or code job needs an artifact with an exact version` | `bad_value` | a `function` or `code` job without `artifact`, or a name or version outside the contract's pattern (a range) |--| `net policy not supported` | `unsupported` | `net` is present and not `deny` |++| `net policy not supported` | `unsupported` | `net` is present and not one of `deny`, `none`, `public`, `openvibe-only`, or is one the host cannot enforce |++| `invalid inputs` | `bad_value` | an entry of `inputs` outside the contract's patterns, a repeated name, or more than 32 |++| `job inputs not supported` | `unsupported` | the job carries `inputs`: this Node cannot stage and check them yet, so it never runs it with them dropped || `missing or invalid args, ttl_ms or limits` | `bad_value` | `args` is not an object, or `ttl_ms` or a limit is missing or below 1 || `class not available` | `unsupported` | the class is valid but this Node does not run it (every class but `function`; every class while the worker is off) || `the e-stop or local stop is latched` | `estopped` or `local_stop` | the stop latch is set: no job starts until it clears |@@@ -90,9 +90,13 @@ the server. The worker is **off by default**; off, the Node refuses every job `cWireGuard meshes), loopback, link-local (`169.254/16`, cloud metadata included), multicast, documentation andreserved ranges, `egress_deny`, the host itself (every address it holds) and anything not routed through thehost's default route (a WireGuard peer, a container bridge) are refused; there is no IPv6 (`fe80::/10` included).-- `openvibe-only`: `public`, further restricted to `egress_allow`. A job runs under the **stricter** of its own `net`-- and this: a job naming `net: "deny"` gets `none` whatever `egress` says; a job without `net` gets `egress`. A policy-- the host cannot enforce fails the probe and keeps the worker off: never a silent `none` nor an open network.++ `openvibe-only`: `public`, further restricted to `egress_allow`. A job runs under the **stricter** of the `net` the++ contract's job names and this. `net` absent means `deny` (`platform.job@1`'s declared default), so **a job that names++ no `net` runs with no network at all** however open this is set: the Node never widens the contract's default. A job++ naming `deny` or `none` gets the same; one naming `public` or `openvibe-only` runs under `egress` when that allows++ that much (`public` under `openvibe-only`, say, runs as `public`) and is **refused** with `net policy not supported`++ when `egress` is `none`: it is never run with less than was asked. A policy the host cannot enforce fails the++ probe and keeps the worker off: never a silent `none` nor an open network.- `egress_allow` (with `openvibe-only` only, and required by it): the IPv4 CIDRs a job may reach, the OpenVibenetwork's own ranges. They are configured, not guessed nor resolved from names: the Node trusts no DNS answer forthem. The refused ranges above still win over an entry that overlaps them.@@@ -126,7 +130,9 @@ AppArmor; a Node running as root with `run_as` is not affected.1. **Admit** (before the `ack`): refused unless the stop latch is clear, the artifact is declared and fewer than`max_jobs` jobs run. An id the worker already holds is answered from it (`ack` while it runs, its `job_exit` once-- it ended) and never starts a second process.++ it ended) and never starts a second process. A job carrying `inputs` is refused (`job inputs not supported`) until++ the Node can place them in its working directory and check their size and sha256: it never runs one with the++ requested files and the digest check silently dropped. So is a job naming a `net` this host cannot enforce.2. **Start**: the Node makes the job's cgroup and starts itself again (`/proc/self/exe`) as the **sandbox init**,with its name as its only argument and its spec (the function's command and environment included) in a memfd, sothat no host user reads the environment in its `/proc/<pid>/cmdline` (the Node checks that it shows the name@@@ -171,8 +177,8 @@ In six lines:`artifact_dir` made as it starts and a size-capped `/tmp` of its own: none of the Node's files, its credential orits control socket, nor the host's private keys.4. The network namespace has only a loopback that is down: no IP network, not even to the Node or `localhost`-- (`egress` `none`, or a job naming `net: "deny"`). Under `public` or `openvibe-only` it also holds one veth to the-- host, through which only what that policy allows leaves (see "Egress"); the loopback stays down.++ (`net` absent, `deny` or `none`, whatever `egress` says). Under `public` or `openvibe-only` it also holds one veth to++ the host, through which only what that policy allows leaves (see "Egress"); the loopback stays down.5. Nothing of the Node's environment reaches it.6. CPU, memory, processes and disk I/O are enforced by its cgroup v2 and rlimits; ttl, wall, CPU and output also by aper-job watchdog that SIGKILLs the whole group; `RLIMIT_CPU` backs the CPU cap per process.@@@ -255,6 +261,8 @@ Refused with `nack` (keyed by the job id; the first matching row wins, after the| `message` | `fault_code` | when ||---------------------------------------|----------------------------|---------------------------------------------------------|| `class not available` | `unsupported` | the worker is off (disabled, not Linux, probe failed), or no declared entry has the job's class (`function` or `code`) |++| `job inputs not supported` | `unsupported` | the job carries `inputs`: this Node cannot stage them and check their digests yet, and never runs a job with them silently dropped |++| `net policy not supported` | `unsupported` | the job names a `net` this host cannot enforce (`public` or `openvibe-only` while `egress` is `none`) || `the e-stop or local stop is latched` | `estopped` or `local_stop` | the stop latch is set || `class not available` | `shutting_down` | the Node is stopping || `unknown artifact` | `unsupported` | no declared entry of the job's class has that name and exact version |@@@ -127,8 +127,10 @@ type WorkerConfig struct {// Egress is the most a job may reach on the network, all three enforced (docs/worker.md): none (the default; a// network namespace with only a loopback that is down), public (IPv4 to public addresses, through a veth the Node// NATs; private, shared, link-local and reserved ranges, EgressDeny and the host itself refused) or openvibe-only-- // (public restricted to EgressAllow). A job runs under the stricter of its own net and this: net "deny" is none-- // whatever Egress says. A policy the host cannot enforce fails the probe, so the worker stays off.++ // (public restricted to EgressAllow). A job runs under the stricter of the net it names and this: a job naming no++ // net (platform.job@1's declared default, deny), or deny or none, gets no network whatever this says; a job naming++ // public or openvibe-only is refused when this is none, never run with less than was asked. A policy the host cannot++ // enforce fails the probe, so the worker stays off.Egress string `json:"egress,omitempty"`// EgressAllow is the IPv4 CIDRs an openvibe-only job may reach (the OpenVibe network's own ranges, configured// rather than guessed); required with openvibe-only and refused with any other policy.@@@ -148,6 +150,24 @@ const (EgressOpenVibeOnly = "openvibe-only")++// StrictestEgress returns the stricter of two worker.egress policies: openvibe-only is stricter than public, public++// than none. An unknown value ranks as none, the strictest that is still enforceable.++func StrictestEgress(a, b string) string {++ rank := func(p string) int {++ switch p {++ case EgressPublic:++ return 1++ case EgressOpenVibeOnly:++ return 2++ }++ return 0++ }++ if rank(a) <= rank(b) {++ return a++ }++ return b++}++// FunctionConfig is one artifact a job may run: Command is started with the job's args as JSON on stdin.type FunctionConfig struct {Name string `json:"name"`@@@ -68,6 +68,12 @@ func TestJobsRefused(t *testing.T) {j.Net = "allow"e.jobNack(j, protocol.JobNetUnsupported)++ // A malformed input is nacked by Check, before the class is even looked at.++ j = testJob(jobID(8))++ j.Inputs = []protocol.JobInput{{Name: "../escape", MediaID: "med_01JAB2C3D4E5F6G7H8J9K0MNPQ",++ SHA256: "9f86d081884c7d659a2feb15b0b4f8f1c0e6ad1d7e6fd2e1b0b6fd1c4f1d2a3b"}}++ e.jobNack(j, protocol.JobBadInputs)++e.jobNack(testJob("job_123"), protocol.JobBadID)j = testJob(jobID(4))@@@ -231,6 +231,19 @@ func TestCodeJobRefusedByFunctionWorker(t *testing.T) {}}++// TestJobWithInputsRefusedOverTheLink: a well formed job carrying inputs reaches the worker and is nacked job inputs++// not supported: the Node cannot stage them yet, and it never acks a job whose files it would silently drop.++func TestJobWithInputsRefusedOverTheLink(t *testing.T) {++ e := startWorkerNode(t)++ j := testJob(jobID(1))++ j.Artifact = &protocol.Artifact{Name: "hello", Version: "1.0.0"}++ j.Inputs = []protocol.JobInput{{Name: "clip.mp4", MediaID: "med_01JAB2C3D4E5F6G7H8J9K0MNPQ",++ SHA256: "9f86d081884c7d659a2feb15b0b4f8f1c0e6ad1d7e6fd2e1b0b6fd1c4f1d2a3b"}}++ if nk := e.jobNack(j, protocol.JobInputsRefused); nk.FaultCode != protocol.FaultUnsupported {++ t.Fatalf("a job with inputs: %+v", nk)++ }++}++// TestJobRunsOverTheLink: a job sent by the fake Bot is acked, then job_started, job_stdout, job_usage (second 0)// and job_exit arrive in that order; a job frame resent while it runs is acked again and starts no second process;// after the end a resent job is answered with the same job_exit, before and after job_exit_ack.@@@ -90,8 +90,14 @@ const (// RuntimeClasses lists every class the contract names; a worker runs only those it advertises.var RuntimeClasses = []string{ClassFunction, ClassCode, ClassBrowser, ClassLinux, ClassDesktop, ClassGPU}--// NetDeny is the only job network policy (and the default when net is absent).--const NetDeny = "deny"++// Job network policies (platform.job@1 net). A job that names none runs under NetDeny: the contract's declared++// default, which is never widened by the Node's own worker.egress.++const (++ NetDeny = "deny" // no network at all; also what an absent net means++ NetNone = "none" // NetDeny under the name the Node's worker.egress uses++ NetPublic = "public" // public IPv4, filtered by the host++ NetOpenVibeOnly = "openvibe-only" // public, restricted to the host's egress_allow++)// CapWorker is the status.capabilities key a worker advertises its runtime classes under (WorkerCapabilities). It is// present only when the Node runs at least one class.@@@ -118,6 +124,8 @@ const (JobUnknownClass = "unknown class"JobNoArtifact = "function or code job needs an artifact with an exact version"JobNetUnsupported = "net policy not supported"++ JobBadInputs = "invalid inputs"++ JobInputsRefused = "job inputs not supported"JobBadLimits = "missing or invalid args, ttl_ms or limits"JobNotAvailable = "class not available"@@@ -280,6 +288,16 @@ type Artifact struct {Version string `json:"version"`}++// JobInput is one file the job's working directory must hold before the process starts: an OpenVibe.Media object++// pinned by digest. The worker fetches it with its own token and never a caller-chosen URL, and a size or digest++// mismatch ends the job failed before the process ever runs.++type JobInput struct {++ Name string `json:"name"`++ MediaID string `json:"media_id"`++ SHA256 string `json:"sha256"`++ SizeBytes int64 `json:"size_bytes,omitempty"`++}++// Job is platform.job@1.type Job struct {ID string `json:"id"` // job_<ULID>, minted by the dispatcher: the idempotency and ack key@@@ -289,6 +307,7 @@ type Job struct {TTLMS int64 `json:"ttl_ms"`Limits JobLimits `json:"limits"`Net string `json:"net,omitempty"` // absent means deny++ Inputs []JobInput `json:"inputs,omitempty"`}// JobRequest is the `job` frame: run Job. A body that does not decode still yields the frame, with Err set and the@@@ -324,11 +343,36 @@ var (jobIDRe = regexp.MustCompile(`^job_[0-9A-HJKMNP-TV-Z]{26}$`)artifactRe = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]{0,127}$`)artifactVerRe = regexp.MustCompile(`^[0-9A-Za-z][0-9A-Za-z.+-]{0,63}$`)++ mediaIDRe = regexp.MustCompile(`^med_[0-9A-HJKMNP-TV-Z]{26}$`)++ sha256Re = regexp.MustCompile(`^[0-9a-f]{64}$`)++ inputNameRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`))// ValidJobID reports whether id is a job_<ULID>.func ValidJobID(id string) bool { return jobIDRe.MatchString(id) }++// ValidJobInputs reports whether the job's inputs are well formed: at most maxJobInputs, each a Media object named by++// its contract's patterns, with names unique and carrying no path separator. An empty list is valid; nil is, too.++func ValidJobInputs(in []JobInput) bool {++ if len(in) > maxJobInputs {++ return false++ }++ seen := make(map[string]bool, len(in))++ for _, i := range in {++ if !inputNameRe.MatchString(i.Name) || !mediaIDRe.MatchString(i.MediaID) || !sha256Re.MatchString(i.SHA256) || i.SizeBytes < 0 {++ return false++ }++ if seen[i.Name] {++ return false++ }++ seen[i.Name] = true++ }++ return true++}++++// maxJobInputs is the contract's cap on a job's inputs array.++const maxJobInputs = 32++// Check returns the fault code and reason to nack this job with, or two empty strings when a worker running the// classes in available takes it. The checks run in this order, so the reason names the first problem.func (r JobRequest) Check(available []string) (fault, reason string) {@@@ -342,8 +386,10 @@ func (r JobRequest) Check(available []string) (fault, reason string) {return FaultUnsupported, JobUnknownClasscase (j.Class == ClassFunction || j.Class == ClassCode) && (j.Artifact == nil || !artifactRe.MatchString(j.Artifact.Name) || !artifactVerRe.MatchString(j.Artifact.Version)):return FaultBadValue, JobNoArtifact-- case j.Net != "" && j.Net != NetDeny:++ case j.Net != "" && j.Net != NetDeny && j.Net != NetNone && j.Net != NetPublic && j.Net != NetOpenVibeOnly:return FaultUnsupported, JobNetUnsupported++ case !ValidJobInputs(j.Inputs):++ return FaultBadValue, JobBadInputscase !isObject(j.Args) || j.TTLMS < 1 || j.Limits.WallMS < 1 || j.Limits.CPUMS < 1 || j.Limits.MemBytes < 1:return FaultBadValue, JobBadLimitscase !contains(available, j.Class):@@@ -3,6 +3,7 @@ package protocolimport ("encoding/json""errors"++ "fmt""os""path/filepath""reflect"@@@ -264,6 +265,16 @@ func TestWorkerCapabilityNames(t *testing.T) {}}++// A well formed input, the Media object and digest the contract's example carries.++const (++ mediaID = "med_01JAB2C3D4E5F6G7H8J9K0MNPQ"++ digest = "9f86d081884c7d659a2feb15b0b4f8f1c0e6ad1d7e6fd2e1b0b6fd1c4f1d2a3b"++)++++func goodInputs() []JobInput {++ return []JobInput{{Name: "clip.mp4", MediaID: mediaID, SHA256: digest, SizeBytes: 1048576}}++}++func TestJobCheck(t *testing.T) {const id = "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"valid := func() Job {@@@ -296,6 +307,24 @@ func TestJobCheck(t *testing.T) {{"version span", func(j *Job) { j.Artifact.Version = "1.2 - 2" }, []string{ClassFunction}, JobNoArtifact},{"bad artifact name", func(j *Job) { j.Artifact.Name = "Thumbnail" }, []string{ClassFunction}, JobNoArtifact},{"net allow", func(j *Job) { j.Net = "allow" }, []string{ClassFunction}, JobNetUnsupported},++ {"net none", func(j *Job) { j.Net = NetNone }, []string{ClassFunction}, ""},++ {"net public", func(j *Job) { j.Net = NetPublic }, []string{ClassFunction}, ""},++ {"net openvibe-only", func(j *Job) { j.Net = NetOpenVibeOnly }, []string{ClassFunction}, ""},++ {"inputs taken", func(j *Job) { j.Inputs = goodInputs() }, []string{ClassFunction}, ""},++ {"input without a digest", func(j *Job) { j.Inputs = []JobInput{{Name: "a.bin", MediaID: mediaID}} }, []string{ClassFunction}, JobBadInputs},++ {"input with a path", func(j *Job) { j.Inputs = []JobInput{{Name: "a/b.bin", MediaID: mediaID, SHA256: digest}} }, []string{ClassFunction}, JobBadInputs},++ {"input twice under one name", func(j *Job) {++ j.Inputs = []JobInput{{Name: "a.bin", MediaID: mediaID, SHA256: digest}, {Name: "a.bin", MediaID: mediaID, SHA256: digest}}++ }, []string{ClassFunction}, JobBadInputs},++ {"input with a negative size", func(j *Job) {++ j.Inputs = []JobInput{{Name: "a.bin", MediaID: mediaID, SHA256: digest, SizeBytes: -1}}++ }, []string{ClassFunction}, JobBadInputs},++ {"too many inputs", func(j *Job) {++ j.Inputs = nil++ for i := 0; i < 33; i++ {++ j.Inputs = append(j.Inputs, JobInput{Name: fmt.Sprintf("f%02d.bin", i), MediaID: mediaID, SHA256: digest})++ }++ }, []string{ClassFunction}, JobBadInputs},{"no args", func(j *Job) { j.Args = nil }, []string{ClassFunction}, JobBadLimits},{"args not an object", func(j *Job) { j.Args = json.RawMessage(`[1]`) }, []string{ClassFunction}, JobBadLimits},{"no ttl", func(j *Job) { j.TTLMS = 0 }, []string{ClassFunction}, JobBadLimits},@@@ -310,6 +339,33 @@ func TestJobCheck(t *testing.T) {}}++// TestJobInputsDecode: the contract's second example (a code job with inputs and net public) decodes to the inputs it++// names, which Check then takes; the same body without inputs keeps none.++func TestJobInputsDecode(t *testing.T) {++ const body = `{"v":1,"seq":3,"ts":1790935200000,"type":"job","job":{"id":"job_01JAB2C3D4E5F6G7H8J9K0MNPR","class":"code",++ "artifact":{"name":"transcode","version":"0.4.1"},"args":{"format":"webm"},"ttl_ms":120000,++ "limits":{"wall_ms":90000,"cpu_ms":90000,"mem_bytes":536870912},"net":"public",++ "inputs":[{"name":"clip.mp4","media_id":"med_01JAB2C3D4E5F6G7H8J9K0MNPQ","sha256":"9f86d081884c7d659a2feb15b0b4f8f1c0e6ad1d7e6fd2e1b0b6fd1c4f1d2a3b","size_bytes":1048576}]}}`++ f, err := Decode([]byte(body))++ if err != nil {++ t.Fatal(err)++ }++ r := f.Msg.(JobRequest)++ if r.Err != nil || r.Job.Net != NetPublic || !reflect.DeepEqual(r.Job.Inputs, goodInputs()) {++ t.Fatalf("%+v", r)++ }++ if fault, reason := r.Check([]string{ClassCode}); fault != "" {++ t.Fatalf("the contract's example is refused: %s %s", fault, reason)++ }++ f, err = Decode([]byte(`{"v":1,"seq":3,"ts":1,"type":"job","job":{"id":"job_01JAB2C3D4E5F6G7H8J9K0MNPR","class":"code","artifact":{"name":"transcode","version":"0.4.1"},"args":{},"ttl_ms":1,"limits":{"wall_ms":1,"cpu_ms":1,"mem_bytes":1}}}`))++ if err != nil {++ t.Fatal(err)++ }++ if j := f.Msg.(JobRequest).Job; len(j.Inputs) != 0 || j.Net != "" {++ t.Fatalf("%+v: inputs and net must stay absent when the body carries none", j)++ }++}++// TestJobUndecodableBody: a job body of the wrong shape still decodes as a job frame, with the id and Err, so the// Node nacks it instead of dropping it.func TestJobUndecodableBody(t *testing.T) {@@@ -1263,8 +1263,9 @@ func TestEgressUnenforceable(t *testing.T) {// TestEgress: under none a job has no route at all; under public it has one, but 10.0.0.1, 169.254.169.254 (cloud// metadata) and the host's end of its veth are refused by the Node's table (refused, not unreachable nor a timeout);--// under openvibe-only an unlisted public host is refused too; a job naming net "deny" under public gets none. It needs--// root, ip, nsenter, nft and IPv4 forwarding: it skips without the programs even where OPENVIBE_WORKER_TESTS=require.++// under openvibe-only an unlisted public host is refused too; a job naming net "deny" or no net at all under public++// gets none whatever the host allows, and one naming public under openvibe-only gets public. It needs root, ip,++// nsenter, nft and IPv4 forwarding: it skips without the programs even where OPENVIBE_WORKER_TESTS=require.func TestEgress(t *testing.T) {none := map[string]string{"10.0.0.1:80": "unreachable", "169.254.169.254:80": "unreachable","169.254.240.1:22": "unreachable", "1.1.1.1:53": "unreachable"}@@@ -1276,10 +1277,14 @@ func TestEgress(t *testing.T) {want map[string]string}{{"none", "", nil, "", none},-- {"public", config.EgressPublic, nil, "", refused},-- {"openvibe-only", config.EgressOpenVibeOnly, []string{"9.9.9.9/32"}, "",++ {"public", config.EgressPublic, nil, protocol.NetPublic, refused},++ {"openvibe-only", config.EgressOpenVibeOnly, []string{"9.9.9.9/32"}, protocol.NetOpenVibeOnly,map[string]string{"10.0.0.1:80": "refused", "169.254.169.254:80": "refused", "1.1.1.1:53": "refused"}},{"deny under public", config.EgressPublic, nil, protocol.NetDeny, none},++ {"no net under public", config.EgressPublic, nil, "", none},++ {"none under openvibe-only", config.EgressOpenVibeOnly, []string{"9.9.9.9/32"}, "", none},++ // A job asking for less than the host allows runs under what it asked.++ {"public under openvibe-only", config.EgressOpenVibeOnly, []string{"9.9.9.9/32"}, protocol.NetPublic, refused},} {t.Run(c.name, func(t *testing.T) {cfg := helperConfig(t, config.WorkerCaps{}, "egress")@@@ -63,14 +63,25 @@ type plan struct {egress string // the job's network policy (egressFor); "" is none}--// egressFor is the network policy a job runs under: the stricter of its net and the host's worker.egress. A job--// naming net "deny" (the only value it may name) runs under none whatever the host allows; without net, the host's--// policy applies.--func egressFor(host, net string) string {-- if net != "" || host == config.EgressNone {-- return ""-- }-- return host++// egressFor is the network policy a job runs under: the stricter of the net the job names and the host's worker.egress.++// A net of deny (or none, the same policy as the Node spells it), which is also what an absent net means in++// platform.job@1, runs under no network whatever the host allows: the contract's default is deny and the Node never++// widens it. A named public or openvibe-only still needs the host to allow that much, so a host under none refuses++// the job at Admit rather than running it with less than was asked.++func egressFor(host, net string) (string, bool) {++ switch net {++ case "", protocol.NetDeny, protocol.NetNone:++ return "", true++ case protocol.NetPublic, protocol.NetOpenVibeOnly:++ if host == config.EgressNone {++ return "", false++ }++ if host == net {++ return host, true++ }++ return config.StrictestEgress(host, net), true++ }++ return "", false}type job struct {@@@ -117,6 +128,14 @@ func (w *Worker) Admit(j protocol.Job) (fault, reason string) {if j.Class != protocol.ClassFunction && j.Class != protocol.ClassCode {return protocol.FaultUnsupported, protocol.JobNotAvailable}++ // inputs are refused until the Node can stage them and check their digests (platform.job@1): a job never runs++ // with the files and the digest check silently dropped.++ if len(j.Inputs) > 0 {++ return protocol.FaultUnsupported, protocol.JobInputsRefused++ }++ if _, ok := egressFor(w.cfg.Egress, j.Net); !ok {++ return protocol.FaultUnsupported, protocol.JobNetUnsupported++ }fn, ok := w.entry(j.Class, j.Artifact)if !ok {return protocol.FaultUnsupported, protocol.JobUnknownArtifact@@@ -402,7 +421,8 @@ func (w *Worker) execute(jb *job) protocol.JobExit {cmd.Dir, cmd.Env = work, jobEnv(jb.fn, work, id)cmd.Stdin = bytes.NewReader(append(append([]byte(nil), jb.req.Args...), '\n'))cmd.Stdout, cmd.ExtraFiles = outW, []*os.File{resW}-- sb, err := w.isolate(cmd, w.cfg, plan{root: dir, fn: jb.fn, limits: jb.limits, egress: egressFor(w.cfg.Egress, jb.req.Net)})++ egress, _ := egressFor(w.cfg.Egress, jb.req.Net) // Admit refused what the host cannot enforce++ sb, err := w.isolate(cmd, w.cfg, plan{root: dir, fn: jb.fn, limits: jb.limits, egress: egress})if err != nil {outW.Close()resW.Close()@@@ -280,8 +280,14 @@ func probe(cfg config.WorkerConfig, isolate isolateFunc) error {if err := policy(cfg); err != nil {return err}-- egress := egressFor(cfg.Egress, "")-- if egress != "" {++ // The host's own policy, the ceiling every job may be offered: a job naming deny or none (or no net at all) runs++ // under no network whatever it says, and one naming public or openvibe-only runs under this, which the probe must be++ // able to enforce (Admit refuses what it cannot).++ egress := cfg.Egress++ if egress == "" {++ egress = config.EgressNone++ }++ if egress != config.EgressNone {if _, err := egressReady(); err != nil {return fmt.Errorf("worker.egress %s: %w", egress, err)}@@@ -289,16 +289,49 @@ func TestUsageHelperProcess(t *testing.T) {os.Exit(0)}--// TestEgressFor: a job runs under the stricter of its net and the host's egress: net "deny" is none whatever the host--// allows; without net, the host's policy.++// TestEgressFor: an absent net (the contract's declared deny) is no network whatever the host allows; net deny and++// none are that too; a named public or openvibe-only runs under the stricter of itself and the host's, and is refused++// when the host is none; a net this Node does not know is refused, never run weaker.func TestEgressFor(t *testing.T) {-- for _, c := range []struct{ host, net, want string }{-- {"", "", ""}, {config.EgressNone, "", ""}, {config.EgressPublic, "", config.EgressPublic},-- {config.EgressOpenVibeOnly, "", config.EgressOpenVibeOnly}, {config.EgressPublic, protocol.NetDeny, ""},-- {config.EgressOpenVibeOnly, protocol.NetDeny, ""}, {config.EgressPublic, "allow", ""},++ for _, c := range []struct {++ host, net string++ want string++ ok bool++ }{++ {"", "", "", true}, {config.EgressNone, "", "", true}, {config.EgressPublic, "", "", true},++ {config.EgressOpenVibeOnly, "", "", true},++ {config.EgressPublic, protocol.NetDeny, "", true}, {config.EgressOpenVibeOnly, protocol.NetNone, "", true},++ {config.EgressPublic, protocol.NetPublic, config.EgressPublic, true},++ {config.EgressPublic, protocol.NetOpenVibeOnly, config.EgressPublic, true},++ {config.EgressOpenVibeOnly, protocol.NetPublic, config.EgressPublic, true},++ {config.EgressOpenVibeOnly, protocol.NetOpenVibeOnly, config.EgressOpenVibeOnly, true},++ {config.EgressNone, protocol.NetPublic, "", false},++ {config.EgressNone, protocol.NetOpenVibeOnly, "", false},++ {config.EgressPublic, "allow", "", false}, {config.EgressPublic, "", "", true},} {-- if got := egressFor(c.host, c.net); got != c.want {-- t.Errorf("egressFor(%q, %q) = %q, want %q", c.host, c.net, got, c.want)++ got, ok := egressFor(c.host, c.net)++ if got != c.want || ok != c.ok {++ t.Errorf("egressFor(%q, %q) = %q, %v; want %q, %v", c.host, c.net, got, ok, c.want, c.ok)}}}++++// TestAdmitRefusesInputs: a job carrying inputs is refused, never run with them silently dropped (platform.job@1).++func TestAdmitRefusesInputs(t *testing.T) {++ s := &sink{}++ w := New(helperConfig(t, config.WorkerCaps{}, "sleep"), s.send, quiet())++ j := testJob(1, "sleep")++ j.Inputs = []protocol.JobInput{{Name: "clip.mp4", MediaID: "med_01JAB2C3D4E5F6G7H8J9K0MNPQ",++ SHA256: "9f86d081884c7d659a2feb15b0b4f8f1c0e6ad1d7e6fd2e1b0b6fd1c4f1d2a3b"}}++ if f, r := w.Admit(j); f != protocol.FaultUnsupported || r != protocol.JobInputsRefused {++ t.Fatalf("a job with inputs: %q %q, want %q %q", f, r, protocol.FaultUnsupported, protocol.JobInputsRefused)++ }++ // A net the host cannot enforce is refused too, rather than run with no network.++ n := testJob(2, "sleep")++ n.Net = protocol.NetPublic++ cfg := helperConfig(t, config.WorkerCaps{}, "sleep")++ cfg.Egress = config.EgressNone++ if f, r := New(cfg, s.send, quiet()).Admit(n); f != protocol.FaultUnsupported || r != protocol.JobNetUnsupported {++ t.Fatalf("net public under egress none: %q %q", f, r)++ }++}
@@ -237,7 +237,7 @@ Otherwise it advertises no class and refuses every well-formed job with `class n
@@ -237,7 +237,7 @@ Otherwise it advertises no class and refuses every well-formed job with `class n
| direction | type | fields |
| direction | type | fields |
|------------------|----------------|-------------------------------------------------------------------------------------------|
|------------------|----------------|-------------------------------------------------------------------------------------------|
-| server → device | `job` | `job` {`id` (`job_<ULID>`: the idempotency and ack key), `class`, `artifact` {`name`, `version`} (required for `function` and `code`, one exact version), `args` {}, `ttl_ms`, `limits` {`wall_ms`, `cpu_ms`, `mem_bytes`}, `net` (`deny`: no network whatever the Node's `worker.egress`; absent: `worker.egress` applies, `none` by default)} |
+| server → device | `job` | `job` {`id` (`job_<ULID>`: the idempotency and ack key), `class`, `artifact` {`name`, `version`} (required for `function` and `code`, one exact version), `args` {}, `ttl_ms`, `limits` {`wall_ms`, `cpu_ms`, `mem_bytes`}, `net` (`deny`, `none`, `public`, `openvibe-only`; absent means `deny`, which the Node never widens to its own `worker.egress`; `public` and `openvibe-only` run under the stricter of what is asked and what `worker.egress` allows, and are refused when it is `none`), `inputs` (files pinned by Media id and sha256: **refused**, `job inputs not supported`, until the Node can stage and verify them)} |
| server → device | `job_cancel` | `id` |
| server → device | `job_cancel` | `id` |
| server → device | `job_exit_ack` | `id` |
| server → device | `job_exit_ack` | `id` |
| device → server | `job_started` | `id`, `started_ms` |
| device → server | `job_started` | `id`, `started_ms` |
@@ -254,7 +254,9 @@ checked in this order:
@@ -254,7 +254,9 @@ checked in this order:
| `malformed job` | `bad_frame` | the `job` object does not decode (a field of the wrong type) |
| `malformed job` | `bad_frame` | the `job` object does not decode (a field of the wrong type) |
| `unknown class` | `unsupported` | `class` is not one of `function`, `code`, `browser`, `linux`, `desktop`, `gpu` |
| `unknown class` | `unsupported` | `class` is not one of `function`, `code`, `browser`, `linux`, `desktop`, `gpu` |
| `function or code job needs an artifact with an exact version` | `bad_value` | a `function` or `code` job without `artifact`, or a name or version outside the contract's pattern (a range) |
| `function or code job needs an artifact with an exact version` | `bad_value` | a `function` or `code` job without `artifact`, or a name or version outside the contract's pattern (a range) |
-| `net policy not supported` | `unsupported` | `net` is present and not `deny` |
+| `net policy not supported` | `unsupported` | `net` is present and not one of `deny`, `none`, `public`, `openvibe-only`, or is one the host cannot enforce |
+| `invalid inputs` | `bad_value` | an entry of `inputs` outside the contract's patterns, a repeated name, or more than 32 |
+| `job inputs not supported` | `unsupported` | the job carries `inputs`: this Node cannot stage and check them yet, so it never runs it with them dropped |
| `missing or invalid args, ttl_ms or limits` | `bad_value` | `args` is not an object, or `ttl_ms` or a limit is missing or below 1 |
| `missing or invalid args, ttl_ms or limits` | `bad_value` | `args` is not an object, or `ttl_ms` or a limit is missing or below 1 |
| `class not available` | `unsupported` | the class is valid but this Node does not run it (every class but `function`; every class while the worker is off) |
| `class not available` | `unsupported` | the class is valid but this Node does not run it (every class but `function`; every class while the worker is off) |
| `the e-stop or local stop is latched` | `estopped` or `local_stop` | the stop latch is set: no job starts until it clears |
| `the e-stop or local stop is latched` | `estopped` or `local_stop` | the stop latch is set: no job starts until it clears |
@@ -90,9 +90,13 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
@@ -90,9 +90,13 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
WireGuard meshes), loopback, link-local (`169.254/16`, cloud metadata included), multicast, documentation and
WireGuard meshes), loopback, link-local (`169.254/16`, cloud metadata included), multicast, documentation and
reserved ranges, `egress_deny`, the host itself (every address it holds) and anything not routed through the
reserved ranges, `egress_deny`, the host itself (every address it holds) and anything not routed through the
host's default route (a WireGuard peer, a container bridge) are refused; there is no IPv6 (`fe80::/10` included).
host's default route (a WireGuard peer, a container bridge) are refused; there is no IPv6 (`fe80::/10` included).
- `openvibe-only`: `public`, further restricted to `egress_allow`. A job runs under the **stricter** of its own `net`
- and this: a job naming `net: "deny"` gets `none` whatever `egress` says; a job without `net` gets `egress`. A policy
- the host cannot enforce fails the probe and keeps the worker off: never a silent `none` nor an open network.
+ `openvibe-only`: `public`, further restricted to `egress_allow`. A job runs under the **stricter** of the `net` the
+ contract's job names and this. `net` absent means `deny` (`platform.job@1`'s declared default), so **a job that names
+ no `net` runs with no network at all** however open this is set: the Node never widens the contract's default. A job
+ naming `deny` or `none` gets the same; one naming `public` or `openvibe-only` runs under `egress` when that allows
+ that much (`public` under `openvibe-only`, say, runs as `public`) and is **refused** with `net policy not supported`
+ when `egress` is `none`: it is never run with less than was asked. A policy the host cannot enforce fails the
+ probe and keeps the worker off: never a silent `none` nor an open network.
- `egress_allow` (with `openvibe-only` only, and required by it): the IPv4 CIDRs a job may reach, the OpenVibe
- `egress_allow` (with `openvibe-only` only, and required by it): the IPv4 CIDRs a job may reach, the OpenVibe
network's own ranges. They are configured, not guessed nor resolved from names: the Node trusts no DNS answer for
network's own ranges. They are configured, not guessed nor resolved from names: the Node trusts no DNS answer for
them. The refused ranges above still win over an entry that overlaps them.
them. The refused ranges above still win over an entry that overlaps them.
@@ -126,7 +130,9 @@ AppArmor; a Node running as root with `run_as` is not affected.
@@ -126,7 +130,9 @@ AppArmor; a Node running as root with `run_as` is not affected.
1. **Admit** (before the `ack`): refused unless the stop latch is clear, the artifact is declared and fewer than
1. **Admit** (before the `ack`): refused unless the stop latch is clear, the artifact is declared and fewer than
`max_jobs` jobs run. An id the worker already holds is answered from it (`ack` while it runs, its `job_exit` once
`max_jobs` jobs run. An id the worker already holds is answered from it (`ack` while it runs, its `job_exit` once
- it ended) and never starts a second process.
+ it ended) and never starts a second process. A job carrying `inputs` is refused (`job inputs not supported`) until
+ the Node can place them in its working directory and check their size and sha256: it never runs one with the
+ requested files and the digest check silently dropped. So is a job naming a `net` this host cannot enforce.
2. **Start**: the Node makes the job's cgroup and starts itself again (`/proc/self/exe`) as the **sandbox init**,
2. **Start**: the Node makes the job's cgroup and starts itself again (`/proc/self/exe`) as the **sandbox init**,
with its name as its only argument and its spec (the function's command and environment included) in a memfd, so
with its name as its only argument and its spec (the function's command and environment included) in a memfd, so
that no host user reads the environment in its `/proc/<pid>/cmdline` (the Node checks that it shows the name
that no host user reads the environment in its `/proc/<pid>/cmdline` (the Node checks that it shows the name
@@ -171,8 +177,8 @@ In six lines:
@@ -171,8 +177,8 @@ In six lines:
`artifact_dir` made as it starts and a size-capped `/tmp` of its own: none of the Node's files, its credential or
`artifact_dir` made as it starts and a size-capped `/tmp` of its own: none of the Node's files, its credential or
its control socket, nor the host's private keys.
its control socket, nor the host's private keys.
4. The network namespace has only a loopback that is down: no IP network, not even to the Node or `localhost`
4. The network namespace has only a loopback that is down: no IP network, not even to the Node or `localhost`
- (`egress` `none`, or a job naming `net: "deny"`). Under `public` or `openvibe-only` it also holds one veth to the
- host, through which only what that policy allows leaves (see "Egress"); the loopback stays down.
+ (`net` absent, `deny` or `none`, whatever `egress` says). Under `public` or `openvibe-only` it also holds one veth to
+ the host, through which only what that policy allows leaves (see "Egress"); the loopback stays down.
5. Nothing of the Node's environment reaches it.
5. Nothing of the Node's environment reaches it.
6. CPU, memory, processes and disk I/O are enforced by its cgroup v2 and rlimits; ttl, wall, CPU and output also by a
6. CPU, memory, processes and disk I/O are enforced by its cgroup v2 and rlimits; ttl, wall, CPU and output also by a
per-job watchdog that SIGKILLs the whole group; `RLIMIT_CPU` backs the CPU cap per process.
per-job watchdog that SIGKILLs the whole group; `RLIMIT_CPU` backs the CPU cap per process.
@@ -255,6 +261,8 @@ Refused with `nack` (keyed by the job id; the first matching row wins, after the
@@ -255,6 +261,8 @@ Refused with `nack` (keyed by the job id; the first matching row wins, after the
| `message` | `fault_code` | when |
| `message` | `fault_code` | when |
|---------------------------------------|----------------------------|---------------------------------------------------------|
|---------------------------------------|----------------------------|---------------------------------------------------------|
| `class not available` | `unsupported` | the worker is off (disabled, not Linux, probe failed), or no declared entry has the job's class (`function` or `code`) |
| `class not available` | `unsupported` | the worker is off (disabled, not Linux, probe failed), or no declared entry has the job's class (`function` or `code`) |
+| `job inputs not supported` | `unsupported` | the job carries `inputs`: this Node cannot stage them and check their digests yet, and never runs a job with them silently dropped |
+| `net policy not supported` | `unsupported` | the job names a `net` this host cannot enforce (`public` or `openvibe-only` while `egress` is `none`) |
| `the e-stop or local stop is latched` | `estopped` or `local_stop` | the stop latch is set |
| `the e-stop or local stop is latched` | `estopped` or `local_stop` | the stop latch is set |
| `class not available` | `shutting_down` | the Node is stopping |
| `class not available` | `shutting_down` | the Node is stopping |
| `unknown artifact` | `unsupported` | no declared entry of the job's class has that name and exact version |
| `unknown artifact` | `unsupported` | no declared entry of the job's class has that name and exact version |
@@ -127,8 +127,10 @@ type WorkerConfig struct {
@@ -127,8 +127,10 @@ type WorkerConfig struct {
// Egress is the most a job may reach on the network, all three enforced (docs/worker.md): none (the default; a
// Egress is the most a job may reach on the network, all three enforced (docs/worker.md): none (the default; a
// network namespace with only a loopback that is down), public (IPv4 to public addresses, through a veth the Node
// network namespace with only a loopback that is down), public (IPv4 to public addresses, through a veth the Node
// NATs; private, shared, link-local and reserved ranges, EgressDeny and the host itself refused) or openvibe-only
// NATs; private, shared, link-local and reserved ranges, EgressDeny and the host itself refused) or openvibe-only
- // (public restricted to EgressAllow). A job runs under the stricter of its own net and this: net "deny" is none
- // whatever Egress says. A policy the host cannot enforce fails the probe, so the worker stays off.
+ // (public restricted to EgressAllow). A job runs under the stricter of the net it names and this: a job naming no
+ // net (platform.job@1's declared default, deny), or deny or none, gets no network whatever this says; a job naming
+ // public or openvibe-only is refused when this is none, never run with less than was asked. A policy the host cannot
+ // enforce fails the probe, so the worker stays off.
Egress string `json:"egress,omitempty"`
Egress string `json:"egress,omitempty"`
// EgressAllow is the IPv4 CIDRs an openvibe-only job may reach (the OpenVibe network's own ranges, configured
// EgressAllow is the IPv4 CIDRs an openvibe-only job may reach (the OpenVibe network's own ranges, configured
// rather than guessed); required with openvibe-only and refused with any other policy.
// rather than guessed); required with openvibe-only and refused with any other policy.
@@ -148,6 +150,24 @@ const (
@@ -148,6 +150,24 @@ const (
EgressOpenVibeOnly = "openvibe-only"
EgressOpenVibeOnly = "openvibe-only"
)
)
+// StrictestEgress returns the stricter of two worker.egress policies: openvibe-only is stricter than public, public
+// than none. An unknown value ranks as none, the strictest that is still enforceable.
+func StrictestEgress(a, b string) string {
+ rank := func(p string) int {
+ switch p {
+ case EgressPublic:
+ return 1
+ case EgressOpenVibeOnly:
+ return 2
+ }
+ return 0
+ }
+ if rank(a) <= rank(b) {
+ return a
+ }
+ return b
+}
+
// FunctionConfig is one artifact a job may run: Command is started with the job's args as JSON on stdin.
// FunctionConfig is one artifact a job may run: Command is started with the job's args as JSON on stdin.
type FunctionConfig struct {
type FunctionConfig struct {
Name string `json:"name"`
Name string `json:"name"`
@@ -68,6 +68,12 @@ func TestJobsRefused(t *testing.T) {
@@ -68,6 +68,12 @@ func TestJobsRefused(t *testing.T) {
j.Net = "allow"
j.Net = "allow"
e.jobNack(j, protocol.JobNetUnsupported)
e.jobNack(j, protocol.JobNetUnsupported)
+ // A malformed input is nacked by Check, before the class is even looked at.
+ j = testJob(jobID(8))
+ j.Inputs = []protocol.JobInput{{Name: "../escape", MediaID: "med_01JAB2C3D4E5F6G7H8J9K0MNPQ",
+ SHA256: "9f86d081884c7d659a2feb15b0b4f8f1c0e6ad1d7e6fd2e1b0b6fd1c4f1d2a3b"}}
+ e.jobNack(j, protocol.JobBadInputs)
+
e.jobNack(testJob("job_123"), protocol.JobBadID)
e.jobNack(testJob("job_123"), protocol.JobBadID)
j = testJob(jobID(4))
j = testJob(jobID(4))
@@ -231,6 +231,19 @@ func TestCodeJobRefusedByFunctionWorker(t *testing.T) {
@@ -231,6 +231,19 @@ func TestCodeJobRefusedByFunctionWorker(t *testing.T) {
}
}
}
}
+// TestJobWithInputsRefusedOverTheLink: a well formed job carrying inputs reaches the worker and is nacked job inputs
+// not supported: the Node cannot stage them yet, and it never acks a job whose files it would silently drop.
+func TestJobWithInputsRefusedOverTheLink(t *testing.T) {
+ e := startWorkerNode(t)
+ j := testJob(jobID(1))
+ j.Artifact = &protocol.Artifact{Name: "hello", Version: "1.0.0"}
+ j.Inputs = []protocol.JobInput{{Name: "clip.mp4", MediaID: "med_01JAB2C3D4E5F6G7H8J9K0MNPQ",
+ SHA256: "9f86d081884c7d659a2feb15b0b4f8f1c0e6ad1d7e6fd2e1b0b6fd1c4f1d2a3b"}}
+ if nk := e.jobNack(j, protocol.JobInputsRefused); nk.FaultCode != protocol.FaultUnsupported {
+ t.Fatalf("a job with inputs: %+v", nk)
+ }
+}
+
// TestJobRunsOverTheLink: a job sent by the fake Bot is acked, then job_started, job_stdout, job_usage (second 0)
// TestJobRunsOverTheLink: a job sent by the fake Bot is acked, then job_started, job_stdout, job_usage (second 0)
// and job_exit arrive in that order; a job frame resent while it runs is acked again and starts no second process;
// and job_exit arrive in that order; a job frame resent while it runs is acked again and starts no second process;
// after the end a resent job is answered with the same job_exit, before and after job_exit_ack.
// after the end a resent job is answered with the same job_exit, before and after job_exit_ack.
@@ -90,8 +90,14 @@ const (
@@ -90,8 +90,14 @@ const (
// RuntimeClasses lists every class the contract names; a worker runs only those it advertises.
// RuntimeClasses lists every class the contract names; a worker runs only those it advertises.
var RuntimeClasses = []string{ClassFunction, ClassCode, ClassBrowser, ClassLinux, ClassDesktop, ClassGPU}
var RuntimeClasses = []string{ClassFunction, ClassCode, ClassBrowser, ClassLinux, ClassDesktop, ClassGPU}
-// NetDeny is the only job network policy (and the default when net is absent).
-const NetDeny = "deny"
+// Job network policies (platform.job@1 net). A job that names none runs under NetDeny: the contract's declared
+// default, which is never widened by the Node's own worker.egress.
+const (
+ NetDeny = "deny" // no network at all; also what an absent net means
+ NetNone = "none" // NetDeny under the name the Node's worker.egress uses
+ NetPublic = "public" // public IPv4, filtered by the host
+ NetOpenVibeOnly = "openvibe-only" // public, restricted to the host's egress_allow
+)
// CapWorker is the status.capabilities key a worker advertises its runtime classes under (WorkerCapabilities). It is
// CapWorker is the status.capabilities key a worker advertises its runtime classes under (WorkerCapabilities). It is
// present only when the Node runs at least one class.
// present only when the Node runs at least one class.
@@ -118,6 +124,8 @@ const (
@@ -118,6 +124,8 @@ const (
JobUnknownClass = "unknown class"
JobUnknownClass = "unknown class"
JobNoArtifact = "function or code job needs an artifact with an exact version"
JobNoArtifact = "function or code job needs an artifact with an exact version"
JobNetUnsupported = "net policy not supported"
JobNetUnsupported = "net policy not supported"
+ JobBadInputs = "invalid inputs"
+ JobInputsRefused = "job inputs not supported"
JobBadLimits = "missing or invalid args, ttl_ms or limits"
JobBadLimits = "missing or invalid args, ttl_ms or limits"
JobNotAvailable = "class not available"
JobNotAvailable = "class not available"
@@ -280,6 +288,16 @@ type Artifact struct {
@@ -280,6 +288,16 @@ type Artifact struct {
Version string `json:"version"`
Version string `json:"version"`
}
}
+// JobInput is one file the job's working directory must hold before the process starts: an OpenVibe.Media object
+// pinned by digest. The worker fetches it with its own token and never a caller-chosen URL, and a size or digest
+// mismatch ends the job failed before the process ever runs.
+type JobInput struct {
+ Name string `json:"name"`
+ MediaID string `json:"media_id"`
+ SHA256 string `json:"sha256"`
+ SizeBytes int64 `json:"size_bytes,omitempty"`
+}
+
// Job is platform.job@1.
// Job is platform.job@1.
type Job struct {
type Job struct {
ID string `json:"id"` // job_<ULID>, minted by the dispatcher: the idempotency and ack key
ID string `json:"id"` // job_<ULID>, minted by the dispatcher: the idempotency and ack key
@@ -289,6 +307,7 @@ type Job struct {
@@ -289,6 +307,7 @@ type Job struct {
TTLMS int64 `json:"ttl_ms"`
TTLMS int64 `json:"ttl_ms"`
Limits JobLimits `json:"limits"`
Limits JobLimits `json:"limits"`
Net string `json:"net,omitempty"` // absent means deny
Net string `json:"net,omitempty"` // absent means deny
+ Inputs []JobInput `json:"inputs,omitempty"`
}
}
// JobRequest is the `job` frame: run Job. A body that does not decode still yields the frame, with Err set and the
// JobRequest is the `job` frame: run Job. A body that does not decode still yields the frame, with Err set and the
@@ -324,11 +343,36 @@ var (
@@ -324,11 +343,36 @@ var (
jobIDRe = regexp.MustCompile(`^job_[0-9A-HJKMNP-TV-Z]{26}$`)
jobIDRe = regexp.MustCompile(`^job_[0-9A-HJKMNP-TV-Z]{26}$`)
artifactRe = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]{0,127}$`)
artifactRe = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]{0,127}$`)
artifactVerRe = regexp.MustCompile(`^[0-9A-Za-z][0-9A-Za-z.+-]{0,63}$`)
artifactVerRe = regexp.MustCompile(`^[0-9A-Za-z][0-9A-Za-z.+-]{0,63}$`)
+ mediaIDRe = regexp.MustCompile(`^med_[0-9A-HJKMNP-TV-Z]{26}$`)
+ sha256Re = regexp.MustCompile(`^[0-9a-f]{64}$`)
+ inputNameRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`)
)
)
// ValidJobID reports whether id is a job_<ULID>.
// ValidJobID reports whether id is a job_<ULID>.
func ValidJobID(id string) bool { return jobIDRe.MatchString(id) }
func ValidJobID(id string) bool { return jobIDRe.MatchString(id) }
+// ValidJobInputs reports whether the job's inputs are well formed: at most maxJobInputs, each a Media object named by
+// its contract's patterns, with names unique and carrying no path separator. An empty list is valid; nil is, too.
+func ValidJobInputs(in []JobInput) bool {
+ if len(in) > maxJobInputs {
+ return false
+ }
+ seen := make(map[string]bool, len(in))
+ for _, i := range in {
+ if !inputNameRe.MatchString(i.Name) || !mediaIDRe.MatchString(i.MediaID) || !sha256Re.MatchString(i.SHA256) || i.SizeBytes < 0 {
+ return false
+ }
+ if seen[i.Name] {
+ return false
+ }
+ seen[i.Name] = true
+ }
+ return true
+}
+
+// maxJobInputs is the contract's cap on a job's inputs array.
+const maxJobInputs = 32
+
// Check returns the fault code and reason to nack this job with, or two empty strings when a worker running the
// Check returns the fault code and reason to nack this job with, or two empty strings when a worker running the
// classes in available takes it. The checks run in this order, so the reason names the first problem.
// classes in available takes it. The checks run in this order, so the reason names the first problem.
func (r JobRequest) Check(available []string) (fault, reason string) {
func (r JobRequest) Check(available []string) (fault, reason string) {
@@ -342,8 +386,10 @@ func (r JobRequest) Check(available []string) (fault, reason string) {
@@ -342,8 +386,10 @@ func (r JobRequest) Check(available []string) (fault, reason string) {
return FaultUnsupported, JobUnknownClass
return FaultUnsupported, JobUnknownClass
case (j.Class == ClassFunction || j.Class == ClassCode) && (j.Artifact == nil || !artifactRe.MatchString(j.Artifact.Name) || !artifactVerRe.MatchString(j.Artifact.Version)):
case (j.Class == ClassFunction || j.Class == ClassCode) && (j.Artifact == nil || !artifactRe.MatchString(j.Artifact.Name) || !artifactVerRe.MatchString(j.Artifact.Version)):
return FaultBadValue, JobNoArtifact
return FaultBadValue, JobNoArtifact
- case j.Net != "" && j.Net != NetDeny:
+ case j.Net != "" && j.Net != NetDeny && j.Net != NetNone && j.Net != NetPublic && j.Net != NetOpenVibeOnly:
return FaultUnsupported, JobNetUnsupported
return FaultUnsupported, JobNetUnsupported
+ case !ValidJobInputs(j.Inputs):
+ return FaultBadValue, JobBadInputs
case !isObject(j.Args) || j.TTLMS < 1 || j.Limits.WallMS < 1 || j.Limits.CPUMS < 1 || j.Limits.MemBytes < 1:
case !isObject(j.Args) || j.TTLMS < 1 || j.Limits.WallMS < 1 || j.Limits.CPUMS < 1 || j.Limits.MemBytes < 1:
return FaultBadValue, JobBadLimits
return FaultBadValue, JobBadLimits
case !contains(available, j.Class):
case !contains(available, j.Class):
@@ -3,6 +3,7 @@ package protocol
@@ -3,6 +3,7 @@ package protocol
import (
import (
"encoding/json"
"encoding/json"
"errors"
"errors"
+ "fmt"
"os"
"os"
"path/filepath"
"path/filepath"
"reflect"
"reflect"
@@ -264,6 +265,16 @@ func TestWorkerCapabilityNames(t *testing.T) {
@@ -264,6 +265,16 @@ func TestWorkerCapabilityNames(t *testing.T) {
}
}
}
}
+// A well formed input, the Media object and digest the contract's example carries.
+const (
+ mediaID = "med_01JAB2C3D4E5F6G7H8J9K0MNPQ"
+ digest = "9f86d081884c7d659a2feb15b0b4f8f1c0e6ad1d7e6fd2e1b0b6fd1c4f1d2a3b"
+)
+
+func goodInputs() []JobInput {
+ return []JobInput{{Name: "clip.mp4", MediaID: mediaID, SHA256: digest, SizeBytes: 1048576}}
+}
+
func TestJobCheck(t *testing.T) {
func TestJobCheck(t *testing.T) {
const id = "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"
const id = "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"
valid := func() Job {
valid := func() Job {
@@ -296,6 +307,24 @@ func TestJobCheck(t *testing.T) {
@@ -296,6 +307,24 @@ func TestJobCheck(t *testing.T) {
{"version span", func(j *Job) { j.Artifact.Version = "1.2 - 2" }, []string{ClassFunction}, JobNoArtifact},
{"version span", func(j *Job) { j.Artifact.Version = "1.2 - 2" }, []string{ClassFunction}, JobNoArtifact},
{"bad artifact name", func(j *Job) { j.Artifact.Name = "Thumbnail" }, []string{ClassFunction}, JobNoArtifact},
{"bad artifact name", func(j *Job) { j.Artifact.Name = "Thumbnail" }, []string{ClassFunction}, JobNoArtifact},
{"net allow", func(j *Job) { j.Net = "allow" }, []string{ClassFunction}, JobNetUnsupported},
{"net allow", func(j *Job) { j.Net = "allow" }, []string{ClassFunction}, JobNetUnsupported},
+ {"net none", func(j *Job) { j.Net = NetNone }, []string{ClassFunction}, ""},
+ {"net public", func(j *Job) { j.Net = NetPublic }, []string{ClassFunction}, ""},
+ {"net openvibe-only", func(j *Job) { j.Net = NetOpenVibeOnly }, []string{ClassFunction}, ""},
+ {"inputs taken", func(j *Job) { j.Inputs = goodInputs() }, []string{ClassFunction}, ""},
+ {"input without a digest", func(j *Job) { j.Inputs = []JobInput{{Name: "a.bin", MediaID: mediaID}} }, []string{ClassFunction}, JobBadInputs},
+ {"input with a path", func(j *Job) { j.Inputs = []JobInput{{Name: "a/b.bin", MediaID: mediaID, SHA256: digest}} }, []string{ClassFunction}, JobBadInputs},
+ {"input twice under one name", func(j *Job) {
+ j.Inputs = []JobInput{{Name: "a.bin", MediaID: mediaID, SHA256: digest}, {Name: "a.bin", MediaID: mediaID, SHA256: digest}}
+ }, []string{ClassFunction}, JobBadInputs},
+ {"input with a negative size", func(j *Job) {
+ j.Inputs = []JobInput{{Name: "a.bin", MediaID: mediaID, SHA256: digest, SizeBytes: -1}}
+ }, []string{ClassFunction}, JobBadInputs},
+ {"too many inputs", func(j *Job) {
+ j.Inputs = nil
+ for i := 0; i < 33; i++ {
+ j.Inputs = append(j.Inputs, JobInput{Name: fmt.Sprintf("f%02d.bin", i), MediaID: mediaID, SHA256: digest})
+ }
+ }, []string{ClassFunction}, JobBadInputs},
{"no args", func(j *Job) { j.Args = nil }, []string{ClassFunction}, JobBadLimits},
{"no args", func(j *Job) { j.Args = nil }, []string{ClassFunction}, JobBadLimits},
{"args not an object", func(j *Job) { j.Args = json.RawMessage(`[1]`) }, []string{ClassFunction}, JobBadLimits},
{"args not an object", func(j *Job) { j.Args = json.RawMessage(`[1]`) }, []string{ClassFunction}, JobBadLimits},
{"no ttl", func(j *Job) { j.TTLMS = 0 }, []string{ClassFunction}, JobBadLimits},
{"no ttl", func(j *Job) { j.TTLMS = 0 }, []string{ClassFunction}, JobBadLimits},
@@ -310,6 +339,33 @@ func TestJobCheck(t *testing.T) {
@@ -310,6 +339,33 @@ func TestJobCheck(t *testing.T) {
}
}
}
}
+// TestJobInputsDecode: the contract's second example (a code job with inputs and net public) decodes to the inputs it
+// names, which Check then takes; the same body without inputs keeps none.
+func TestJobInputsDecode(t *testing.T) {
+ const body = `{"v":1,"seq":3,"ts":1790935200000,"type":"job","job":{"id":"job_01JAB2C3D4E5F6G7H8J9K0MNPR","class":"code",
+ "artifact":{"name":"transcode","version":"0.4.1"},"args":{"format":"webm"},"ttl_ms":120000,
+ "limits":{"wall_ms":90000,"cpu_ms":90000,"mem_bytes":536870912},"net":"public",
+ "inputs":[{"name":"clip.mp4","media_id":"med_01JAB2C3D4E5F6G7H8J9K0MNPQ","sha256":"9f86d081884c7d659a2feb15b0b4f8f1c0e6ad1d7e6fd2e1b0b6fd1c4f1d2a3b","size_bytes":1048576}]}}`
+ f, err := Decode([]byte(body))
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := f.Msg.(JobRequest)
+ if r.Err != nil || r.Job.Net != NetPublic || !reflect.DeepEqual(r.Job.Inputs, goodInputs()) {
+ t.Fatalf("%+v", r)
+ }
+ if fault, reason := r.Check([]string{ClassCode}); fault != "" {
+ t.Fatalf("the contract's example is refused: %s %s", fault, reason)
+ }
+ f, err = Decode([]byte(`{"v":1,"seq":3,"ts":1,"type":"job","job":{"id":"job_01JAB2C3D4E5F6G7H8J9K0MNPR","class":"code","artifact":{"name":"transcode","version":"0.4.1"},"args":{},"ttl_ms":1,"limits":{"wall_ms":1,"cpu_ms":1,"mem_bytes":1}}}`))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if j := f.Msg.(JobRequest).Job; len(j.Inputs) != 0 || j.Net != "" {
+ t.Fatalf("%+v: inputs and net must stay absent when the body carries none", j)
+ }
+}
+
// TestJobUndecodableBody: a job body of the wrong shape still decodes as a job frame, with the id and Err, so the
// TestJobUndecodableBody: a job body of the wrong shape still decodes as a job frame, with the id and Err, so the
// Node nacks it instead of dropping it.
// Node nacks it instead of dropping it.
func TestJobUndecodableBody(t *testing.T) {
func TestJobUndecodableBody(t *testing.T) {
@@ -1263,8 +1263,9 @@ func TestEgressUnenforceable(t *testing.T) {
@@ -1263,8 +1263,9 @@ func TestEgressUnenforceable(t *testing.T) {
// TestEgress: under none a job has no route at all; under public it has one, but 10.0.0.1, 169.254.169.254 (cloud
// TestEgress: under none a job has no route at all; under public it has one, but 10.0.0.1, 169.254.169.254 (cloud
// metadata) and the host's end of its veth are refused by the Node's table (refused, not unreachable nor a timeout);
// metadata) and the host's end of its veth are refused by the Node's table (refused, not unreachable nor a timeout);
-// under openvibe-only an unlisted public host is refused too; a job naming net "deny" under public gets none. It needs
-// root, ip, nsenter, nft and IPv4 forwarding: it skips without the programs even where OPENVIBE_WORKER_TESTS=require.
+// under openvibe-only an unlisted public host is refused too; a job naming net "deny" or no net at all under public
+// gets none whatever the host allows, and one naming public under openvibe-only gets public. It needs root, ip,
+// nsenter, nft and IPv4 forwarding: it skips without the programs even where OPENVIBE_WORKER_TESTS=require.
func TestEgress(t *testing.T) {
func TestEgress(t *testing.T) {
none := map[string]string{"10.0.0.1:80": "unreachable", "169.254.169.254:80": "unreachable",
none := map[string]string{"10.0.0.1:80": "unreachable", "169.254.169.254:80": "unreachable",
"169.254.240.1:22": "unreachable", "1.1.1.1:53": "unreachable"}
"169.254.240.1:22": "unreachable", "1.1.1.1:53": "unreachable"}
@@ -1276,10 +1277,14 @@ func TestEgress(t *testing.T) {
@@ -1276,10 +1277,14 @@ func TestEgress(t *testing.T) {
want map[string]string
want map[string]string
}{
}{
{"none", "", nil, "", none},
{"none", "", nil, "", none},
- {"public", config.EgressPublic, nil, "", refused},
- {"openvibe-only", config.EgressOpenVibeOnly, []string{"9.9.9.9/32"}, "",
+ {"public", config.EgressPublic, nil, protocol.NetPublic, refused},
+ {"openvibe-only", config.EgressOpenVibeOnly, []string{"9.9.9.9/32"}, protocol.NetOpenVibeOnly,
map[string]string{"10.0.0.1:80": "refused", "169.254.169.254:80": "refused", "1.1.1.1:53": "refused"}},
map[string]string{"10.0.0.1:80": "refused", "169.254.169.254:80": "refused", "1.1.1.1:53": "refused"}},
{"deny under public", config.EgressPublic, nil, protocol.NetDeny, none},
{"deny under public", config.EgressPublic, nil, protocol.NetDeny, none},
+ {"no net under public", config.EgressPublic, nil, "", none},
+ {"none under openvibe-only", config.EgressOpenVibeOnly, []string{"9.9.9.9/32"}, "", none},
+ // A job asking for less than the host allows runs under what it asked.
+ {"public under openvibe-only", config.EgressOpenVibeOnly, []string{"9.9.9.9/32"}, protocol.NetPublic, refused},
} {
} {
t.Run(c.name, func(t *testing.T) {
t.Run(c.name, func(t *testing.T) {
cfg := helperConfig(t, config.WorkerCaps{}, "egress")
cfg := helperConfig(t, config.WorkerCaps{}, "egress")
@@ -63,14 +63,25 @@ type plan struct {
@@ -63,14 +63,25 @@ type plan struct {
egress string // the job's network policy (egressFor); "" is none
egress string // the job's network policy (egressFor); "" is none
}
}
-// egressFor is the network policy a job runs under: the stricter of its net and the host's worker.egress. A job
-// naming net "deny" (the only value it may name) runs under none whatever the host allows; without net, the host's
-// policy applies.
-func egressFor(host, net string) string {
- if net != "" || host == config.EgressNone {
- return ""
- }
- return host
+// egressFor is the network policy a job runs under: the stricter of the net the job names and the host's worker.egress.
+// A net of deny (or none, the same policy as the Node spells it), which is also what an absent net means in
+// platform.job@1, runs under no network whatever the host allows: the contract's default is deny and the Node never
+// widens it. A named public or openvibe-only still needs the host to allow that much, so a host under none refuses
+// the job at Admit rather than running it with less than was asked.
+func egressFor(host, net string) (string, bool) {
+ switch net {
+ case "", protocol.NetDeny, protocol.NetNone:
+ return "", true
+ case protocol.NetPublic, protocol.NetOpenVibeOnly:
+ if host == config.EgressNone {
+ return "", false
+ }
+ if host == net {
+ return host, true
+ }
+ return config.StrictestEgress(host, net), true
+ }
+ return "", false
}
}
type job struct {
type job struct {
@@ -117,6 +128,14 @@ func (w *Worker) Admit(j protocol.Job) (fault, reason string) {
@@ -117,6 +128,14 @@ func (w *Worker) Admit(j protocol.Job) (fault, reason string) {
if j.Class != protocol.ClassFunction && j.Class != protocol.ClassCode {
if j.Class != protocol.ClassFunction && j.Class != protocol.ClassCode {
return protocol.FaultUnsupported, protocol.JobNotAvailable
return protocol.FaultUnsupported, protocol.JobNotAvailable
}
}
+ // inputs are refused until the Node can stage them and check their digests (platform.job@1): a job never runs
+ // with the files and the digest check silently dropped.
+ if len(j.Inputs) > 0 {
+ return protocol.FaultUnsupported, protocol.JobInputsRefused
+ }
+ if _, ok := egressFor(w.cfg.Egress, j.Net); !ok {
+ return protocol.FaultUnsupported, protocol.JobNetUnsupported
+ }
fn, ok := w.entry(j.Class, j.Artifact)
fn, ok := w.entry(j.Class, j.Artifact)
if !ok {
if !ok {
return protocol.FaultUnsupported, protocol.JobUnknownArtifact
return protocol.FaultUnsupported, protocol.JobUnknownArtifact
@@ -402,7 +421,8 @@ func (w *Worker) execute(jb *job) protocol.JobExit {
@@ -402,7 +421,8 @@ func (w *Worker) execute(jb *job) protocol.JobExit {
cmd.Dir, cmd.Env = work, jobEnv(jb.fn, work, id)
cmd.Dir, cmd.Env = work, jobEnv(jb.fn, work, id)
cmd.Stdin = bytes.NewReader(append(append([]byte(nil), jb.req.Args...), '\n'))
cmd.Stdin = bytes.NewReader(append(append([]byte(nil), jb.req.Args...), '\n'))
cmd.Stdout, cmd.ExtraFiles = outW, []*os.File{resW}
cmd.Stdout, cmd.ExtraFiles = outW, []*os.File{resW}
- sb, err := w.isolate(cmd, w.cfg, plan{root: dir, fn: jb.fn, limits: jb.limits, egress: egressFor(w.cfg.Egress, jb.req.Net)})
+ egress, _ := egressFor(w.cfg.Egress, jb.req.Net) // Admit refused what the host cannot enforce
+ sb, err := w.isolate(cmd, w.cfg, plan{root: dir, fn: jb.fn, limits: jb.limits, egress: egress})
if err != nil {
if err != nil {
outW.Close()
outW.Close()
resW.Close()
resW.Close()
@@ -280,8 +280,14 @@ func probe(cfg config.WorkerConfig, isolate isolateFunc) error {
@@ -280,8 +280,14 @@ func probe(cfg config.WorkerConfig, isolate isolateFunc) error {
if err := policy(cfg); err != nil {
if err := policy(cfg); err != nil {
return err
return err
}
}
- egress := egressFor(cfg.Egress, "")
- if egress != "" {
+ // The host's own policy, the ceiling every job may be offered: a job naming deny or none (or no net at all) runs
+ // under no network whatever it says, and one naming public or openvibe-only runs under this, which the probe must be
+ // able to enforce (Admit refuses what it cannot).
+ egress := cfg.Egress
+ if egress == "" {
+ egress = config.EgressNone
+ }
+ if egress != config.EgressNone {
if _, err := egressReady(); err != nil {
if _, err := egressReady(); err != nil {
return fmt.Errorf("worker.egress %s: %w", egress, err)
return fmt.Errorf("worker.egress %s: %w", egress, err)
}
}
@@ -289,16 +289,49 @@ func TestUsageHelperProcess(t *testing.T) {
@@ -289,16 +289,49 @@ func TestUsageHelperProcess(t *testing.T) {
os.Exit(0)
os.Exit(0)
}
}
-// TestEgressFor: a job runs under the stricter of its net and the host's egress: net "deny" is none whatever the host
-// allows; without net, the host's policy.
+// TestEgressFor: an absent net (the contract's declared deny) is no network whatever the host allows; net deny and
+// none are that too; a named public or openvibe-only runs under the stricter of itself and the host's, and is refused
+// when the host is none; a net this Node does not know is refused, never run weaker.
func TestEgressFor(t *testing.T) {
func TestEgressFor(t *testing.T) {
- for _, c := range []struct{ host, net, want string }{
- {"", "", ""}, {config.EgressNone, "", ""}, {config.EgressPublic, "", config.EgressPublic},
- {config.EgressOpenVibeOnly, "", config.EgressOpenVibeOnly}, {config.EgressPublic, protocol.NetDeny, ""},
- {config.EgressOpenVibeOnly, protocol.NetDeny, ""}, {config.EgressPublic, "allow", ""},
+ for _, c := range []struct {
+ host, net string
+ want string
+ ok bool
+ }{
+ {"", "", "", true}, {config.EgressNone, "", "", true}, {config.EgressPublic, "", "", true},
+ {config.EgressOpenVibeOnly, "", "", true},
+ {config.EgressPublic, protocol.NetDeny, "", true}, {config.EgressOpenVibeOnly, protocol.NetNone, "", true},
+ {config.EgressPublic, protocol.NetPublic, config.EgressPublic, true},
+ {config.EgressPublic, protocol.NetOpenVibeOnly, config.EgressPublic, true},
+ {config.EgressOpenVibeOnly, protocol.NetPublic, config.EgressPublic, true},
+ {config.EgressOpenVibeOnly, protocol.NetOpenVibeOnly, config.EgressOpenVibeOnly, true},
+ {config.EgressNone, protocol.NetPublic, "", false},
+ {config.EgressNone, protocol.NetOpenVibeOnly, "", false},
+ {config.EgressPublic, "allow", "", false}, {config.EgressPublic, "", "", true},
} {
} {
- if got := egressFor(c.host, c.net); got != c.want {
- t.Errorf("egressFor(%q, %q) = %q, want %q", c.host, c.net, got, c.want)
+ got, ok := egressFor(c.host, c.net)
+ if got != c.want || ok != c.ok {
+ t.Errorf("egressFor(%q, %q) = %q, %v; want %q, %v", c.host, c.net, got, ok, c.want, c.ok)
}
}
}
}
}
}
+
+// TestAdmitRefusesInputs: a job carrying inputs is refused, never run with them silently dropped (platform.job@1).
+func TestAdmitRefusesInputs(t *testing.T) {
+ s := &sink{}
+ w := New(helperConfig(t, config.WorkerCaps{}, "sleep"), s.send, quiet())
+ j := testJob(1, "sleep")
+ j.Inputs = []protocol.JobInput{{Name: "clip.mp4", MediaID: "med_01JAB2C3D4E5F6G7H8J9K0MNPQ",
+ SHA256: "9f86d081884c7d659a2feb15b0b4f8f1c0e6ad1d7e6fd2e1b0b6fd1c4f1d2a3b"}}
+ if f, r := w.Admit(j); f != protocol.FaultUnsupported || r != protocol.JobInputsRefused {
+ t.Fatalf("a job with inputs: %q %q, want %q %q", f, r, protocol.FaultUnsupported, protocol.JobInputsRefused)
+ }
+ // A net the host cannot enforce is refused too, rather than run with no network.
+ n := testJob(2, "sleep")
+ n.Net = protocol.NetPublic
+ cfg := helperConfig(t, config.WorkerCaps{}, "sleep")
+ cfg.Egress = config.EgressNone
+ if f, r := New(cfg, s.send, quiet()).Admit(n); f != protocol.FaultUnsupported || r != protocol.JobNetUnsupported {
+ t.Fatalf("net public under egress none: %q %q", f, r)
+ }
+}