[scheduler]: ACLs phase 1 - per Job ACL specification and enforcement.

scheduler/appengine/engine/engine.go

 // Copyright 2015 The LUCI Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //      http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ skipping 30 matching lines @@
         "github.com/luci/luci-go/common/data/rand/mathrand"
         "github.com/luci/luci-go/common/data/stringset"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/retry/transient"
         "github.com/luci/luci-go/server/auth"
         "github.com/luci/luci-go/server/auth/identity"
         "github.com/luci/luci-go/server/auth/signing"
         "github.com/luci/luci-go/server/tokens"
 
+        "github.com/luci/luci-go/scheduler/appengine/acl"
         "github.com/luci/luci-go/scheduler/appengine/catalog"
         "github.com/luci/luci-go/scheduler/appengine/schedule"
         "github.com/luci/luci-go/scheduler/appengine/task"
 )
 
 var (
         ErrNoSuchJob        = errors.New("no such job")
         ErrNoSuchInvocation = errors.New("the invocation doesn't exist")
 )
 
 // Engine manages all scheduler jobs: keeps track of their state, runs state
 // machine transactions, starts new invocations, etc. A method returns
 // errors.Transient if the error is non-fatal and the call should be retried
 // later. Any other error means that retry won't help.
 type Engine interface {
         // GetAllProjects returns a list of all projects that have at least one
         // enabled scheduler job.
+        // ACLs are NOT enforced; internal scheduler use only.
         GetAllProjects(c context.Context) ([]string, error)
 
-        // GetAllJobs returns a list of all enabled scheduler jobs in no particular
-        // order.
-        GetAllJobs(c context.Context) ([]*Job, error)
+        // GetAllJobsRA returns a list of all enabled scheduler jobs to which
+        // current identity has READER access in no particular order.
+        GetAllJobsRA(c context.Context) ([]*Job, error)

[tandrii(chromium), 2017/08/02 20:23:46]

RA here and below stands for READER Access. I wish I can get rid of that prefix,
but GetAllProjects will be an outlier then :( 

 
-        // GetProjectJobs returns a list of enabled scheduler jobs of some project
-        // in no particular order.
-        GetProjectJobs(c context.Context, projectID string) ([]*Job, error)
+        // GetProjectJobsRA returns a list of enabled scheduler jobs to which curreent
+        // identity has READER access of some project in no particular order.
+        GetProjectJobsRA(c context.Context, projectID string) ([]*Job, error)
 
-        // GetJob returns single scheduler job given its full ID or nil if no such
-        // job.
-        GetJob(c context.Context, jobID string) (*Job, error)
+        // GetJobRA returns single scheduler job given its full ID if and only if
+        // current identity has READER access to it. Returns nil otherwise.
+        GetJobRA(c context.Context, jobID string) (*Job, error)
 
         // ListInvocations returns invocations of a job, most recent first.
         // Returns fetched invocations and cursor string if there's more.
+        // TODO(tandrii): enforce ACLs.
         ListInvocations(c context.Context, jobID string, pageSize int, cursor string) ([]*Invocation, string, error)

[tandrii(chromium), 2017/08/02 20:23:46]

this and two methods below will fetch JobId concurrently (or sequentially for
...byNonce) to ensure ACLs. Internal engine will call lowercase versions. SGTM?

[Vadim Sh., 2017/08/02 21:14:13]

Yes.

 
         // GetInvocation returns single invocation of some job given its ID.
+        // TODO(tandrii): enforce ACLs.
         GetInvocation(c context.Context, jobID string, invID int64) (*Invocation, error)
 
         // GetInvocationsByNonce returns a list of Invocations with given nonce.
         //
         // Invocation nonce is a random number that identifies an intent to start
         // an invocation. Normally one nonce corresponds to one Invocation entity,
         // but there can be more if job fails to start with a transient error.
+        // TODO(tandrii): enforce ACLs.
         GetInvocationsByNonce(c context.Context, invNonce int64) ([]*Invocation, error)
 
         // UpdateProjectJobs adds new, removes old and updates existing jobs.
         UpdateProjectJobs(c context.Context, projectID string, defs []catalog.Definition) error
 
         // ResetAllJobsOnDevServer forcefully resets state of all enabled jobs.
         // Supposed to be used only on devserver, where task queue stub state is not
         // preserved between appserver restarts and it messes everything.
         ResetAllJobsOnDevServer(c context.Context) error
 
@@ skipping 14 matching lines @@
         // It is needed to be able to manually tests PubSub related workflows on dev
         // server, since dev server can't accept PubSub push messages.
         PullPubSubOnDevServer(c context.Context, taskManagerName, publisher string) error
 
         // TriggerInvocation launches job invocation right now if job isn't running
         // now. Used by "Run now" UI button.
         //
         // Returns new invocation nonce (a random number that identifies an intent to
         // start an invocation). Normally one nonce corresponds to one Invocation
         // entity, but there can be more if job fails to start with a transient error.
         TriggerInvocation(c context.Context, jobID string, triggeredBy identity.Identity) (int64, error)

[tandrii(chromium), 2017/08/02 20:23:46]

this and methods below: i want to remove triggeredBy and instead use whatever is
in context and enforcing OWNER acl in the process. WDYT?

[Vadim Sh., 2017/08/02 21:14:13]

sgtm

 
         // PauseJob replaces job's schedule with "triggered", effectively preventing
         // it from running automatically (until it is resumed). Manual invocations are
         // still allowed. Does nothing if job is already paused. Any pending or
         // running invocations are still executed.
         PauseJob(c context.Context, jobID string, who identity.Identity) error
 
         // ResumeJob resumes paused job. Doesn't nothing if the job is not paused.
         ResumeJob(c context.Context, jobID string, who identity.Identity) error
 
@@ skipping 123 matching lines @@
         // an appropriate revision.
         RevisionURL string `gae:",noindex"`
 
         // Schedule is the job's schedule in regular cron expression format.
         Schedule string `gae:",noindex"`
 
         // Task is the job's payload in serialized form. Opaque from the point of view
         // of the engine. See Catalog.UnmarshalTask().
         Task []byte `gae:",noindex"`
 
+        // ACLs are the latest ACLs applied to Job and all its invocations.
+        Acls acl.GrantsByRole `gae:",noindex"`
+
         // State is the job's state machine state, see StateMachine.
         State JobState
 }
 
 // GetJobName returns name of this Job as defined its project's config.
 func (e *Job) GetJobName() string {
         // JobID has form <project>/<id>. Split it into components.
         chunks := strings.Split(e.JobID, "/")
         return chunks[1]
 }
@@ skipping 39 matching lines @@
                 e.State == other.State)
 }
 
 // matches returns true if job definition in the entity matches the one
 // specified by catalog.Definition struct. UpdateProjectJobs skips updates for
 // such jobs (assuming they are up-to-date).
 func (e *Job) matches(def catalog.Definition) bool {
         return e.JobID == def.JobID &&
                 e.Flavor == def.Flavor &&
                 e.Schedule == def.Schedule &&
+                e.Acls.Equal(&def.Acls) &&
                 bytes.Equal(e.Task, def.Task)
 }
 
 // Invocation entity stores single attempt to run a job. Its parent entity
 // is corresponding Job, its ID is generated based on time.
 type Invocation struct {
         _kind  string         `gae:"$kind,Invocation"`
         _extra ds.PropertyMap `gae:"-,extra"`
 
         // ID is identifier of this particular attempt to run a job. Multiple attempts
@@ skipping 271 matching lines @@
         // Filter out duplicates, sort.
         projects := stringset.New(len(entities))
         for _, ent := range entities {
                 projects.Add(ent.ProjectID)
         }
         out := projects.ToSlice()
         sort.Strings(out)
         return out, nil
 }
 
-func (e *engineImpl) GetAllJobs(c context.Context) ([]*Job, error) {
+func (e *engineImpl) GetAllJobsRA(c context.Context) ([]*Job, error) {
         q := ds.NewQuery("Job").Eq("Enabled", true)
-        return e.queryEnabledJobs(c, q)
+        return e.queryEnabledJobsRA(c, q)
 }
 
-func (e *engineImpl) GetProjectJobs(c context.Context, projectID string) ([]*Job, error) {
+func (e *engineImpl) GetProjectJobsRA(c context.Context, projectID string) ([]*Job, error) {
         q := ds.NewQuery("Job").Eq("Enabled", true).Eq("ProjectID", projectID)
-        return e.queryEnabledJobs(c, q)
+        return e.queryEnabledJobsRA(c, q)
 }
 
-func (e *engineImpl) queryEnabledJobs(c context.Context, q *ds.Query) ([]*Job, error) {
+func (e *engineImpl) queryEnabledJobsRA(c context.Context, q *ds.Query) ([]*Job, error) {
         entities := []*Job{}
         if err := ds.GetAll(c, q, &entities); err != nil {
                 return nil, transient.Tag.Apply(err)
         }
         // Non-ancestor query used, need to recheck filters.
         filtered := make([]*Job, 0, len(entities))
         for _, job := range entities {
-                if job.Enabled {
+                if !job.Enabled {
+                        continue
+                }
+                // TODO(tandrii): improve batch ACLs check here to take advantage of likely
+                // shared ACLs between most jobs of the same project.
+                if ok, err := job.Acls.IsReader(c); err != nil {
+                        return nil, transient.Tag.Apply(err)
+                } else if ok {
                         filtered = append(filtered, job)
                 }
         }
         return filtered, nil
 }
 
-func (e *engineImpl) GetJob(c context.Context, jobID string) (*Job, error) {
+func (e *engineImpl) GetJobRA(c context.Context, jobID string) (*Job, error) {
         job := &Job{JobID: jobID}
         switch err := ds.Get(c, job); {
         case err == nil:
+                if hasAccess, err := job.Acls.IsReader(c); err != nil {
+                        return nil, transient.Tag.Apply(err)
+                } else if !hasAccess {
+                        return nil, nil
+                }
                 return job, nil
         case err == ds.ErrNoSuchEntity:
                 return nil, nil
         default:
                 return nil, transient.Tag.Apply(err)
         }
 }
 
 func (e *engineImpl) ListInvocations(c context.Context, jobID string, pageSize int, cursor string) ([]*Invocation, string, error) {
         if pageSize == 0 || pageSize > 500 {
@@ skipping 1291 matching lines @@
                 }
                 if hasFinished {
                         return ctl.eng.rollSM(c, job, func(sm *StateMachine) error {
                                 sm.OnInvocationDone(saving.ID)
                                 return nil
                         })
                 }
                 return nil
         })
 }