[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")
+        ErrNoOwnerPermission = errors.New("no OWNER permission on a job")
+        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.
+// ACLs are enforced unlike EngineInternal with the following implications:
+//  * if caller lacks have READER access to Jobs, methods behave as if Jobs

[Vadim Sh., 2017/08/04 23:07:26]

"lacks have" ?

[tandrii(chromium), 2017/08/07 12:48:17]

Done.

+//    do not exist.
+//  * if caller lacks have OWNER access, calling mutating methods will result in

[Vadim Sh., 2017/08/04 23:07:27]

same here

[tandrii(chromium), 2017/08/07 12:48:17]

Done.

+//    ErrNoOwnerPermission (assuming caller has READER access, else see above).
 type Engine interface {
+        // GetVisibleProjects returns a list of all projects that have at least one
+        // enabled scheduler job to which caller has READER access.
+        // GetVisibleProjects(c context.Context) ([]string, error)

[Vadim Sh., 2017/08/04 23:07:27]

remove

I assume it's not needed now?

[tandrii(chromium), 2017/08/07 12:48:16]

Yep, Done.

+
+        // GetVisibleJobs returns a list of all enabled scheduler jobs in no
+        // particular order.
+        GetVisibleJobs(c context.Context) ([]*Job, error)
+
+        // GetVisibleProjectJobs returns a list of enabled scheduler jobs of some
+        // project in no particular order.
+        GetVisibleProjectJobs(c context.Context, projectID string) ([]*Job, error)
+
+        // GetVisibleJob returns single scheduler job given its full ID or nil if no such
+        // job or if not visible.
+        GetVisibleJob(c context.Context, jobID string) (*Job, error)
+
+        // ListVisibleInvocations returns invocations of a visible job, most recent first.
+        // Returns fetched invocations and cursor string if there's more.
+        ListVisibleInvocations(c context.Context, jobID string, pageSize int, cursor string) ([]*Invocation, string, error)
+
+        // GetVisibleInvocation returns single invocation of some job given its ID.
+        GetVisibleInvocation(c context.Context, jobID string, invID int64) (*Invocation, error)
+
+        // GetVisibleInvocationsByNonce 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.
+        GetVisibleInvocationsByNonce(c context.Context, invNonce int64) ([]*Invocation, error)
+
+        // 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) error
+
+        // ResumeJob resumes paused job. Does nothing if the job is not paused.
+        ResumeJob(c context.Context, jobID string) error
+
+        // AbortJob resets the job to scheduled state, aborting a currently pending or
+        // running invocation (if any).
+        //
+        // Returns nil if the job is not currently running.
+        AbortJob(c context.Context, jobID string) error
+
+        // AbortInvocation forcefully moves the invocation to failed state.
+        //
+        // It opportunistically tries to send "abort" signal to a job runner if it
+        // supports cancellation, but it doesn't wait for reply. It proceeds to
+        // modifying local state in the scheduler service datastore immediately.
+        //
+        // AbortInvocation can be used to manually "unstuck" jobs that got stuck due
+        // to missing PubSub notifications or other kinds of unexpected conditions.
+        //
+        // Does nothing if invocation is already in some final state.
+        AbortInvocation(c context.Context, jobID string, invID int64) 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) (int64, error)
+}
+
+// EngineInternal is to be used by frontend initialization code only.
+type EngineInternal interface {
         // GetAllProjects returns a list of all projects that have at least one
         // enabled scheduler job.
         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)
-
-        // GetProjectJobs returns a list of enabled scheduler jobs of some project
-        // in no particular order.
-        GetProjectJobs(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)
-
-        // ListInvocations returns invocations of a job, most recent first.
-        // Returns fetched invocations and cursor string if there's more.
-        ListInvocations(c context.Context, jobID string, pageSize int, cursor string) ([]*Invocation, string, error)
-
-        // GetInvocation returns single invocation of some job given its ID.
-        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.
-        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
 
         // ExecuteSerializedAction is called via a task queue to execute an action
         // produced by job state machine transition. These actions are POSTed
         // to TimersQueue and InvocationsQueue defined in Config by Engine.
         // 'retryCount' is 0 on first attempt, 1 if task queue service retries
         // request once, 2 - if twice, and so on. Returning transient errors here
         // causes the task queue to retry the task.
         ExecuteSerializedAction(c context.Context, body []byte, retryCount int) error
 
         // ProcessPubSubPush is called whenever incoming PubSub message is received.
         ProcessPubSubPush(c context.Context, body []byte) error
 
         // PullPubSubOnDevServer is called on dev server to pull messages from PubSub
         // subscription associated with given publisher.
         //
         // 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)
-
-        // 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
-
-        // AbortInvocation forcefully moves the invocation to failed state.
-        //
-        // It opportunistically tries to send "abort" signal to a job runner if it
-        // supports cancellation, but it doesn't wait for reply. It proceeds to
-        // modifying local state in the scheduler service datastore immediately.
-        //
-        // AbortInvocation can be used to manually "unstuck" jobs that got stuck due
-        // to missing PubSub notifications or other kinds of unexpected conditions.
-        //
-        // Does nothing if invocation is already in some final state.
-        AbortInvocation(c context.Context, jobID string, invID int64, who identity.Identity) error
-
-        // AbortJob resets the job to scheduled state, aborting a currently pending or
-        // running invocation (if any).
-        //
-        // Returns nil if the job is not currently running.
-        AbortJob(c context.Context, jobID string, who identity.Identity) error
+        PublicAPI() Engine

[Vadim Sh., 2017/08/04 23:07:27]

document

[tandrii(chromium), 2017/08/07 12:48:17]

Done.

 }
 
 // Config contains parameters for the engine.
 type Config struct {
         Catalog              catalog.Catalog // provides task.Manager's to run tasks
         TimersQueuePath      string          // URL of a task queue handler for timer ticks
         TimersQueueName      string          // queue name for timer ticks
         InvocationsQueuePath string          // URL of a task queue handler that starts jobs
         InvocationsQueueName string          // queue name for job starts
         PubSubPushPath       string          // URL to use in PubSub push config
 }
 
-// NewEngine returns default implementation of Engine.
-func NewEngine(conf Config) Engine {
+// NewEngine returns default implementation of EngineInternal.
+func NewEngine(conf Config) EngineInternal {
         return &engineImpl{
                 Config:    conf,
                 doneFlags: make(map[string]bool),
         }
 }
 
 //// Implementation.
 
 const (
         // invocationRetryLimit is how many times to retry an invocation before giving
@@ skipping 82 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 190 matching lines @@
 
         return 0, errors.New("could not find available invocationID after 10 attempts")
 }
 
 // debugLog mutates a string by appending a line to it.
 func debugLog(c context.Context, str *string, format string, args ...interface{}) {
         prefix := clock.Now(c).UTC().Format("[15:04:05.000] ")
         *str += prefix + fmt.Sprintf(format+"\n", args...)
 }
 
-////
+////////////////////////////////////////////////////////////////////////////////
+// engineImpl.
 
 type engineImpl struct {
         Config
 
         lock      sync.Mutex
         doneFlags map[string]bool // see doIfNotDone
 
         // configureTopic is used by prepareTopic, mocked in tests.
         configureTopic func(c context.Context, topic, sub, pushURL, publisher string) error
 }
@@ skipping 60 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) GetVisibleJobs(c context.Context) ([]*Job, error) {
         q := ds.NewQuery("Job").Eq("Enabled", true)
-        return e.queryEnabledJobs(c, q)
+        return e.queryEnabledVisibleJobs(c, q)
 }
 
-func (e *engineImpl) GetProjectJobs(c context.Context, projectID string) ([]*Job, error) {
+func (e *engineImpl) GetVisibleProjectJobs(c context.Context, projectID string) ([]*Job, error) {
         q := ds.NewQuery("Job").Eq("Enabled", true).Eq("ProjectID", projectID)
-        return e.queryEnabledJobs(c, q)
+        return e.queryEnabledVisibleJobs(c, q)
 }
 
-func (e *engineImpl) queryEnabledJobs(c context.Context, q *ds.Query) ([]*Job, error) {
+func (e *engineImpl) queryEnabledVisibleJobs(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) GetVisibleJob(c context.Context, jobID string) (*Job, error) {
+        job, err := e.getJob(c, jobID)
+        if err != nil {
+                return nil, err
+        } else if job == nil {
+                return nil, ErrNoSuchJob
+        }
+        if ok, err := job.Acls.IsReader(c); err != nil {
+                return nil, err
+        } else if !ok {
+                return nil, ErrNoSuchJob
+        }
+        return job, err

[Vadim Sh., 2017/08/04 23:07:26]

nit: return job, nil

[tandrii(chromium), 2017/08/07 12:48:17]

Done.

+}
+
+func (e *engineImpl) getOwnedJob(c context.Context, jobID string) (*Job, error) {
+        job, err := e.getJob(c, jobID)
+        if err != nil {
+                return nil, err
+        } else if job == nil {
+                return nil, ErrNoSuchJob
+        }
+
+        switch owner, err := job.Acls.IsOwner(c); {
+        case err != nil:
+                return nil, err
+        case owner:
+                return job, nil
+        }
+
+        // Not owner, but maybe reader? Give nicer error in such case.
+        switch reader, err := job.Acls.IsReader(c); {
+        case err != nil:
+                return nil, err
+        case reader:
+                return nil, ErrNoOwnerPermission
+        default:
+                return nil, ErrNoSuchJob
+        }
+}
+
+func (e *engineImpl) getJob(c context.Context, jobID string) (*Job, error) {
         job := &Job{JobID: jobID}
         switch err := ds.Get(c, job); {
         case err == 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) {
+func (e *engineImpl) PublicAPI() Engine {
+        return e
+}
+
+func (e *engineImpl) ListVisibleInvocations(c context.Context, jobID string, pageSize int, cursor string) ([]*Invocation, string, error) {
+        if job, err := e.GetVisibleJob(c, jobID); err != nil {

[Vadim Sh., 2017/08/04 23:07:27]

strictly speaking, it would be better if we put ACLs into separate subentity of
main Job entity (in fact, duplicate it there, keeping it also in Job), to avoid
fetching frequently changing Job entity all the time (experiencing dscache cache
misses).

In practice it probably doesn't matter, QPS here is negligible.

[tandrii(chromium), 2017/08/07 12:48:17]

First, since API and UI QPS is indeed negligible, I'd rather do this later since
IIUC it will still be easily doable once this CL lands.

Second, do IIUC that you suggest to create smth like

type JobACLs struct {
  _kind  string         `gae:"$kind,JobACLs"`
  _extra ds.PropertyMap `gae:"-,extra"`

  // JobKey is the key of parent Job entity.
  JobKey *ds.Key `gae:"$parent"`
 
  Acls acl.GrantsByRole `gae:",noindex"`
}

If so, i think i'll need to update ACLs of a job in a transaction to avoid
inconsistent ACLs, right?

I've added a TODO to consider doing this in the future.

[Vadim Sh., 2017/08/07 22:53:21]

Sure, it was just a general thought. I don't think we need to do it anytime
soon.
Right.

+                return nil, "", err
+        } else if job == nil {
+                // Either no Job or no access, both imply returning no invocations.
+                // return []*Invocation{}, "", nil
+                return nil, "", ErrNoSuchInvocation
+        }
+
         if pageSize == 0 || pageSize > 500 {
                 pageSize = 500
         }
 
         // Deserialize the cursor.
         var cursorObj ds.Cursor
         if cursor != "" {
                 var err error
                 cursorObj, err = ds.DecodeCursor(c, cursor)
                 if err != nil {
@@ skipping 11 matching lines @@
         }
 
         // Fetch pageSize worth of invocations, then grab the cursor.
         out := make([]*Invocation, 0, pageSize)
         var newCursor string
         err := ds.Run(c, q, func(obj *Invocation, getCursor ds.CursorCB) error {
                 out = append(out, obj)
                 if len(out) < pageSize {
                         return nil
                 }
-                c, err := getCursor()
+                cur, err := getCursor()
                 if err != nil {
                         return err
                 }
-                newCursor = c.String()
+                newCursor = cur.String()
                 return ds.Stop
         })
         if err != nil {
                 return nil, "", transient.Tag.Apply(err)
         }
         return out, newCursor, nil
 }
 
-func (e *engineImpl) GetInvocation(c context.Context, jobID string, invID int64) (*Invocation, error) {
+func (e *engineImpl) GetVisibleInvocation(c context.Context, jobID string, invID int64) (*Invocation, error) {
+        switch _, err := e.GetVisibleJob(c, jobID); {
+        case err == ErrNoSuchJob:
+                return nil, ErrNoSuchInvocation
+        case err != nil:
+                return nil, err
+        default:
+                return e.getInvocation(c, jobID, invID)
+        }
+}
+
+func (e *engineImpl) getInvocation(c context.Context, jobID string, invID int64) (*Invocation, error) {
         inv := &Invocation{
                 ID:     invID,
                 JobKey: ds.NewKey(c, "Job", jobID, 0, nil),
         }
         switch err := ds.Get(c, inv); {
         case err == nil:
                 return inv, nil
         case err == ds.ErrNoSuchEntity:
                 return nil, nil
         default:
                 return nil, transient.Tag.Apply(err)
         }
 }
 
-func (e *engineImpl) GetInvocationsByNonce(c context.Context, invNonce int64) ([]*Invocation, error) {
+func (e *engineImpl) GetVisibleInvocationsByNonce(c context.Context, invNonce int64) ([]*Invocation, error) {
         q := ds.NewQuery("Invocation").Eq("InvocationNonce", invNonce)
         entities := []*Invocation{}
         if err := ds.GetAll(c, q, &entities); err != nil {
                 return nil, transient.Tag.Apply(err)
         }
+        if len(entities) > 0 {
+                // All Invocations with the same nonce must belong to the same Job.
+                switch _, err := e.GetVisibleJob(c, entities[0].JobKey.StringID()); {
+                case err == ErrNoSuchJob:
+                        return []*Invocation{}, nil
+                case err != nil:
+                        return nil, err
+                }
+        }
         return entities, nil
 }
 
 func (e *engineImpl) UpdateProjectJobs(c context.Context, projectID string, defs []catalog.Definition) error {
         // JobID -> *Job map.
         existing, err := e.getProjectJobs(c, projectID)
         if err != nil {
                 return err
         }
         // JobID -> new definition revision map.
@@ skipping 309 matching lines @@
 
 func (e *engineImpl) executeInvAction(c context.Context, payload *actionTaskPayload, retryCount int) error {
         switch {
         case payload.InvTimer != nil:
                 return e.invocationTimerTick(c, payload.JobID, payload.InvID, payload.InvTimer)
         default:
                 return fmt.Errorf("unexpected invocation action kind %q", payload)
         }
 }
 
-func (e *engineImpl) TriggerInvocation(c context.Context, jobID string, triggeredBy identity.Identity) (int64, error) {
+func (e *engineImpl) TriggerInvocation(c context.Context, jobID string) (int64, error) {
+        // First, we check ACLs.
+        if _, err := e.getOwnedJob(c, jobID); err != nil {
+                return 0, err
+        }
+
         var err error
         var invNonce int64
         err2 := e.txn(c, jobID, func(c context.Context, job *Job, isNew bool) error {
                 if isNew {
                         err = ErrNoSuchJob
                         return errSkipPut
                 }
                 if !job.Enabled {
                         err = errors.New("the job is disabled")
                         return errSkipPut
                 }
                 invNonce = 0
                 return e.rollSM(c, job, func(sm *StateMachine) error {
-                        if err := sm.OnManualInvocation(triggeredBy); err != nil {
+                        if err := sm.OnManualInvocation(auth.CurrentIdentity(c)); err != nil {
                                 return err
                         }
                         invNonce = sm.State.InvocationNonce
                         return nil
                 })
         })
         if err == nil {
                 err = err2
         }
         return invNonce, err
 }
 
-func (e *engineImpl) PauseJob(c context.Context, jobID string, who identity.Identity) error {
-        return e.setPausedFlag(c, jobID, true, who)
+func (e *engineImpl) PauseJob(c context.Context, jobID string) error {
+        // First, we check ACLs.
+        if _, err := e.getOwnedJob(c, jobID); err != nil {

[Vadim Sh., 2017/08/04 23:07:26]

why not move it into 'setPausedFlag'?

[tandrii(chromium), 2017/08/07 12:48:17]

Good point. Done.

+                return err
+        }
+        return e.setPausedFlag(c, jobID, true, auth.CurrentIdentity(c))
 }
 
-func (e *engineImpl) ResumeJob(c context.Context, jobID string, who identity.Identity) error {
-        return e.setPausedFlag(c, jobID, false, who)
+func (e *engineImpl) ResumeJob(c context.Context, jobID string) error {
+        // First, we check ACLs.
+        if _, err := e.getOwnedJob(c, jobID); err != nil {
+                return err
+        }
+        return e.setPausedFlag(c, jobID, false, auth.CurrentIdentity(c))
 }
 
 func (e *engineImpl) setPausedFlag(c context.Context, jobID string, paused bool, who identity.Identity) error {
         return e.txn(c, jobID, func(c context.Context, job *Job, isNew bool) error {
                 if isNew || !job.Enabled {
                         return ErrNoSuchJob
                 }
                 if job.Paused == paused {
                         return errSkipPut
                 }
                 if paused {
                         logging.Warningf(c, "Job is paused by %s", who)
                 } else {
                         logging.Warningf(c, "Job is resumed by %s", who)
                 }
                 job.Paused = paused
                 return e.rollSM(c, job, func(sm *StateMachine) error {
                         sm.OnScheduleChange()
                         return nil
                 })
         })
 }
 
-func (e *engineImpl) AbortInvocation(c context.Context, jobID string, invID int64, who identity.Identity) error {
+func (e *engineImpl) AbortInvocation(c context.Context, jobID string, invID int64) error {
+        if _, err := e.getOwnedJob(c, jobID); err != nil {
+                return err
+        }
+        return e.abortInvocation(c, jobID, invID)
+}
+
+func (e *engineImpl) abortInvocation(c context.Context, jobID string, invID int64) error {
         c = logging.SetField(c, "JobID", jobID)
         c = logging.SetField(c, "InvID", invID)
 
         var inv *Invocation
         var err error
-        switch inv, err = e.GetInvocation(c, jobID, invID); {
+        switch inv, err = e.getInvocation(c, jobID, invID); {
         case err != nil:
                 logging.Errorf(c, "Failed to fetch the invocation - %s", err)
                 return err
         case inv == nil:
                 logging.Errorf(c, "The invocation doesn't exist")
                 return ErrNoSuchInvocation
         case inv.Status.Final():
                 return nil
         }
 
         ctl, err := e.controllerForInvocation(c, inv)
         if err != nil {
                 logging.Errorf(c, "Cannot get controller - %s", err)
                 return err
         }
 
-        ctl.DebugLog("Invocation is manually aborted by %q", who)
+        ctl.DebugLog("Invocation is manually aborted by %q", auth.CurrentUser(c))
         if err = ctl.manager.AbortTask(c, ctl); err != nil {
                 logging.Errorf(c, "Failed to abort the task - %s", err)
                 return err
         }
 
         ctl.State().Status = task.StatusAborted
         if err = ctl.Save(c); err != nil {
                 logging.Errorf(c, "Failed to save the invocation - %s", err)
                 return err
         }
         return nil
 }
 
 // AbortJob resets the job to scheduled state, aborting a currently pending or
 // running invocation (if any).
 //
 // Returns nil if the job is not currently running.
-func (e *engineImpl) AbortJob(c context.Context, jobID string, who identity.Identity) error {
-        // First we switch the job to the default state and disassociate the running
+func (e *engineImpl) AbortJob(c context.Context, jobID string) error {
+        // First, we check ACLs.
+        if _, err := e.getOwnedJob(c, jobID); err != nil {
+                return err
+        }
+        // Second, we switch the job to the default state and disassociate the running
         // invocation (if any) from the job entity.
         var invID int64
         err := e.txn(c, jobID, func(c context.Context, job *Job, isNew bool) error {
                 if isNew {
                         return errSkipPut // the job was removed, nothing to abort
                 }
                 invID = job.State.InvocationID
                 return e.rollSM(c, job, func(sm *StateMachine) error {
                         sm.OnManualAbort()
                         return nil
                 })
         })
         if err != nil {
                 return err
         }
 
         // Now we kill the invocation. We do it separately because it may involve
         // an RPC to remote service (e.g. to cancel a task) that can't be done from
         // the transaction.
         if invID != 0 {
-                return e.AbortInvocation(c, jobID, invID, who)
+                return e.abortInvocation(c, jobID, invID)
         }
         return nil
 }
 
 // updateJob updates an existing job if its definition has changed, adds
 // a completely new job or enables a previously disabled job.
 func (e *engineImpl) updateJob(c context.Context, def catalog.Definition) error {
         return e.txn(c, def.JobID, func(c context.Context, job *Job, isNew bool) error {
                 if !isNew && job.Enabled && job.matches(def) {
                         return errSkipPut
@@ skipping 126 matching lines @@
 }
 
 // invocationTimerTick is called via Task Queue to handle AddTimer callbacks.
 //
 // See also handlePubSubMessage, it is quite similar.
 func (e *engineImpl) invocationTimerTick(c context.Context, jobID string, invID int64, timer *invocationTimer) error {
         c = logging.SetField(c, "JobID", jobID)
         c = logging.SetField(c, "InvID", invID)
 
         logging.Infof(c, "Handling invocation timer %q", timer.Name)
-        inv, err := e.GetInvocation(c, jobID, invID)
+        inv, err := e.getInvocation(c, jobID, invID)
         if err != nil {
                 logging.Errorf(c, "Failed to fetch the invocation - %s", err)
                 return err
         }
         if inv == nil {
                 return ErrNoSuchInvocation
         }
 
         // Finished invocations are immutable, skip the message.
         if inv.Status.Final() {
@@ skipping 364 matching lines @@
                 return err
         }
         jobID = data["job"]
         if invID, err = strconv.ParseInt(data["inv"], 10, 64); err != nil {
                 logging.Errorf(c, "Could not parse 'inv' %q - %s", data["inv"], err)
                 return err
         }
 
         c = logging.SetField(c, "JobID", jobID)
         c = logging.SetField(c, "InvID", invID)
-        inv, err := e.GetInvocation(c, jobID, invID)
+        inv, err := e.getInvocation(c, jobID, invID)
         if err != nil {
                 logging.Errorf(c, "Failed to fetch the invocation - %s", err)
                 return err
         }
         if inv == nil {
                 return ErrNoSuchInvocation
         }
 
         // Finished invocations are immutable, skip the message.
         if inv.Status.Final() {
@@ skipping 236 matching lines @@
                 }
                 if hasFinished {
                         return ctl.eng.rollSM(c, job, func(sm *StateMachine) error {
                                 sm.OnInvocationDone(saving.ID)
                                 return nil
                         })
                 }
                 return nil
         })
 }