Background jobs, and three edges deliberately not drawn

The first edges in the graph that do not start at an account. They were worth
adding only once `finding_neighbourhood` existed to read them — `path_actors`
walks account-sourced edges alone, so adding these earlier would have been more
edges nothing reads, which is the mistake this whole arc began by fixing.

    edges                    75  ->  78
    runs_as / schedules / runs_program      5 / 3 / 3

`check_job_privileged_step_user` had already written down what it wanted, and
could not say it: "the job->user edge is what makes 'this armed job runs as
DDIC' derivable". Third module in a row where the relationship was designed in a
comment and unexpressible in the contract.

MOST OF THIS CHANGE IS ABOUT THE EDGE NOT DRAWN.

JOBCMD-JOB-005 reports a low-privileged scheduler getting code executed under a
powerful step user. The obvious edge is scheduler -> step user, and it would be
wrong: it reads as "this account can act as that one", which is a conclusion a
reader draws from two facts rather than a relationship the export states. So it
is `user -schedules-> job` and `job -runs_as-> user`, and a test fails if a
user -> user edge is ever declared here. It is the same transitive claim
`why_not_the_transitive_edge` already refuses, wearing a different costume.

JOBCMD-JOB-003's own comment sketches "job -> program -> user", which reads as
though RSBDCOS0 runs as the step user. It does not: the job runs the program AND
the job runs as the user. Two edges from the job, and a test asserts nothing is
declared FROM the program.

`runs_as` points from the job, so it is deliberately NOT an actor edge. A job
borrows an identity; the user does not reach for the job, and answering "who is
standing on this path" with somebody a job impersonates would be a different
claim. `path_actors` leaves it alone; `finding_neighbourhood` surfaces it.

A COUNTER EARNING ITS PLACE. Adding the JOBCMD-JOB- rule pushed
`declined_ambiguous` from 1 to 2, because JOB-003 then matched a rule without
having declared its pairs — visible immediately in the scan output rather than
silently absent. Converting it brought the count back to 1.

5,502 pass with a database. 4 new tests.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: Krishcalin

data/graph_edges.json

@@ -20,6 +20,27 @@
       "edge_type": "holds_profile",
       "why": "USR-002 reports users holding SAP_ALL, SAP_NEW or S_A.SYSTEM, and the assignment IS the finding. A profile is not a role — in SAP a role generates a profile and the two are separate objects — so this is its own edge type rather than a second spelling of holds_role. The check declares the pairing in `relations`, so which user holds which profile is evidence here rather than inference."
     },
+    {
+      "check_prefix": "JOBCMD-JOB-",
+      "from_type": "job",
+      "to_type": "user",
+      "edge_type": "runs_as",
+      "why": "An armed background job executes its steps with the step user's authorisations (TBTCP-AUTHCKNAM). `check_job_privileged_step_user` already said what this is for: 'the job->user edge is what makes \"this armed job runs as DDIC\" derivable'. The direction is the job's, not the user's — the job borrows the identity, the user does not reach for the job — so this is NOT an actor edge and `path_actors` does not walk it."
+    },
+    {
+      "check_prefix": "JOBCMD-JOB-003",
+      "from_type": "job",
+      "to_type": "program",
+      "edge_type": "runs_program",
+      "why": "An armed job step names the ABAP program it executes (TBTCP-PROGNAME), and RSBDCOS0 in that position is the OS bridge that bypasses the SM69 allowlist entirely. Emitted as a second edge FROM THE JOB rather than as the chain the check's own comment sketches: 'job -> program -> user' reads as though the program runs as the step user, and nothing in the export says that — the job runs the program and the job runs as the user."
+    },
+    {
+      "check_prefix": "JOBCMD-JOB-005",
+      "from_type": "user",
+      "to_type": "job",
+      "edge_type": "schedules",
+      "why": "The scheduler named in SDLUNAME arranged for the job to run. Kept separate from runs_as deliberately: together the two edges say a lower-privileged scheduler gets code executed under a stronger identity, which is exactly what the check reports — but as two facts a reader joins, never as a scheduler->step-user edge. That edge would read as 'this account can act as that one', a conclusion rather than something the export states, and the same transitive claim why_not_the_transitive_edge already refuses."
+    },
     {
       "check_prefix": "S4AUTHZ-001",
       "from_type": "user",

modules/basis_job_command.py

@@ -383,6 +383,7 @@ def check_command_any_os(self):
     # --------------------------------------------------------------- JOB checks
     def check_job_privileged_step_user(self):
         crit, high = [], []
+        crit_rels, high_rels = [], []
         crit_objs, high_objs = [], []
         for label, step, _h in self._armed_steps():
             user = self._get(step, "AUTHCKNAM", "STEP_USER", "BTCUSER", "AUTH_USER")
@@ -398,12 +399,18 @@ def check_job_privileged_step_user(self):
                     crit_objs.append({"type": "job", "name": label})
                 if user:
                     crit_objs.append({"type": "user", "name": user})
+                if label and user:
+                    crit_rels.append({"from": {"type": "job", "name": label},
+                                      "to": {"type": "user", "name": user}})
             elif cls == "standard":
                 high.append(f"{label} — step user {user}")
                 if label:
                     high_objs.append({"type": "job", "name": label})
                 if user:
                     high_objs.append({"type": "user", "name": user})
+                if label and user:
+                    high_rels.append({"from": {"type": "job", "name": label},
+                                      "to": {"type": "user", "name": user}})
         if crit:
             self.finding(
                 check_id="JOBCMD-JOB-001",
@@ -422,6 +429,7 @@ def check_job_privileged_step_user(self):
                 # Aggregate: re-planning one job onto a technical user must not retire
                 # this finding and re-raise it with a zeroed age while others remain.
                 affected_objects=crit_objs,
+                relations=crit_rels,
                 scope="aggregate",
                 remediation=(
                     "Reassign these jobs to a dedicated, least-privilege background (type B / "
@@ -447,6 +455,7 @@ def check_job_privileged_step_user(self):
                 affected_items=high,
                 # Aggregate, for the same reason as JOBCMD-JOB-001.
                 affected_objects=high_objs,
+                relations=high_rels,
                 scope="aggregate",
                 remediation=(
                     "Move these jobs to a purpose-built least-privilege background user; lock the "
@@ -519,6 +528,7 @@ def check_job_external_step(self):
     def check_job_os_report(self):
         offenders = []
         objects = []
+        relations = []
         for label, step, _h in self._armed_steps():
             prog = self._get(step, "PROGNAME", "REPORT", "ABAP_PROGRAM").upper()
             user = self._get(step, "AUTHCKNAM", "STEP_USER", "BTCUSER")
@@ -541,6 +551,18 @@ def check_job_os_report(self):
                     objects.append({"type": "program", "name": prog})
                 if user:
                     objects.append({"type": "user", "name": user})
+                # TWO EDGES FROM THE JOB, not the chain the comment above
+                # sketches. "job -> program -> user" reads as though the program
+                # runs as the user; it does not — the JOB runs the program AND
+                # runs as the user, and the program is one of its steps. Drawing
+                # it as a chain would put a relationship between RSBDCOS0 and the
+                # step user that nothing in the export states.
+                if label and prog:
+                    relations.append({"from": {"type": "job", "name": label},
+                                      "to": {"type": "program", "name": prog}})
+                if label and user:
+                    relations.append({"from": {"type": "job", "name": label},
+                                      "to": {"type": "user", "name": user}})
         if offenders:
             self.finding(
                 check_id="JOBCMD-JOB-003",
@@ -558,6 +580,7 @@ def check_job_os_report(self):
                 # Aggregate: unscheduling one RSBDCOS0 job while another remains must not
                 # reset this finding's clock.
                 affected_objects=objects,
+                relations=relations,
                 scope="aggregate",
                 remediation=(
                     "Remove RSBDCOS0 from scheduled jobs and restrict authorization to run it; "
@@ -572,6 +595,7 @@ def check_job_os_report(self):
     def check_job_stale_step_user(self):
         offenders = []
         objects = []
+        relations = []
         for label, step, _h in self._armed_steps():
             user = self._get(step, "AUTHCKNAM", "STEP_USER", "BTCUSER").upper()
             if not user:
@@ -599,6 +623,9 @@ def check_job_stale_step_user(self):
                     objects.append({"type": "job", "name": label})
                 if user:
                     objects.append({"type": "user", "name": user})
+                if label and user:
+                    relations.append({"from": {"type": "job", "name": label},
+                                      "to": {"type": "user", "name": user}})
         if offenders:
             self.finding(
                 check_id="JOBCMD-JOB-004",
@@ -615,6 +642,7 @@ def check_job_stale_step_user(self):
                 affected_items=offenders,
                 # Aggregate: a job-hygiene backlog. Fixing one job step user shortens it.
                 affected_objects=objects,
+                relations=relations,
                 scope="aggregate",
                 remediation=(
                     "Re-plan affected jobs onto a valid, unlocked, least-privilege system (type B) "
@@ -627,6 +655,7 @@ def check_job_stale_step_user(self):
     def check_job_identity_borrow(self):
         offenders = []
         objects = []
+        relations = []
         for label, step, header in self._armed_steps():
             user = self._get(step, "AUTHCKNAM", "STEP_USER", "BTCUSER").upper()
             sched = self._get(step, "SDLUNAME", "SCHEDULER", "CREATOR").upper() or \
@@ -640,6 +669,20 @@ def check_job_identity_borrow(self):
                     objects.append({"type": "job", "name": label})
                 objects.append({"type": "user", "name": user})
                 objects.append({"type": "user", "name": sched})
+                # THE BORROWING, AS TWO STEPS RATHER THAN ONE. The scheduler
+                # schedules the job; the job runs as the step user. Declaring a
+                # scheduler -> step user edge instead would be the transitive
+                # claim `data/graph_edges.json` refuses elsewhere — it reads as
+                # "this account can act as that one", which is the CONCLUSION a
+                # reader should draw for themselves from two facts, not a
+                # relationship the export states.
+                if label:
+                    relations.append(
+                        {"from": {"type": "user", "name": sched},
+                         "to": {"type": "job", "name": label}})
+                    relations.append(
+                        {"from": {"type": "job", "name": label},
+                         "to": {"type": "user", "name": user}})
         if offenders:
             self.finding(
                 check_id="JOBCMD-JOB-005",
@@ -656,6 +699,7 @@ def check_job_identity_borrow(self):
                 # Aggregate: the set of borrowing relationships. Correcting one job's step
                 # user must not orphan the finding's history for the rest.
                 affected_objects=objects,
+                relations=relations,
                 scope="aggregate",
                 remediation=(
                     "Ensure the step user reflects the least privilege actually required and that "

tests/test_declared_relations.py

@@ -330,6 +330,76 @@ def test_a_role_collection_is_not_a_pfcg_role():
     assert rules[("user", "role")] == "holds_role"
 
 
+# --------------------------------------------------------------------------- #
+#  Background jobs: the first edges that do NOT start at an account            #
+# --------------------------------------------------------------------------- #
+
+def jobs(steps, headers=None, **rest):
+    from modules.basis_job_command import BasisJobCommandAuditor
+    data = {"background_job_steps": steps,
+            "background_jobs": headers or [], **rest}
+    return {f["check_id"]: f
+            for f in BasisJobCommandAuditor(data, {}, {}).run_all_checks()}
+
+
+def test_a_job_declares_the_identity_it_runs_as():
+    """`check_job_privileged_step_user` already said what this is for: 'the
+    job->user edge is what makes "this armed job runs as DDIC" derivable'."""
+    got = jobs([{"JOBNAME": "JOB_FI_CLOSE", "JOBCOUNT": "1",
+                 "AUTHCKNAM": "DDIC", "PROGNAME": "ZFI_CLOSE"},
+                {"JOBNAME": "JOB_MASS_UPD", "JOBCOUNT": "2",
+                 "AUTHCKNAM": "SAP*", "PROGNAME": "ZMASS"}])
+    fired = [f for cid, f in got.items()
+             if cid.startswith("JOBCMD-JOB-001") and f.get("relations")]
+    assert fired, "no job finding declared what it runs as"
+    pairs = {(r["from"]["name"], r["to"]["name"])
+             for f in fired for r in f["relations"]}
+    assert ("JOB_FI_CLOSE", "DDIC") in pairs, pairs
+
+
+def test_runs_as_points_from_the_job_and_is_not_an_actor_edge():
+    """The job borrows the identity; the user does not reach for the job. So
+    `path_actors` must not walk it — answering "who is standing here" with
+    somebody a job impersonates would be a different claim entirely."""
+    from server import graph
+    rules = {r["edge_type"]: r for r in load_rules()}
+    assert rules["runs_as"]["from_type"] == "job"
+    assert "runs_as" not in graph._ACTOR_EDGES
+
+
+def test_identity_borrowing_is_two_edges_and_never_one():
+    """A low-privileged scheduler getting code run as a powerful step user is
+    the finding. Declaring scheduler -> step user directly would read as "this
+    account can act as that one" — a conclusion the reader should draw from two
+    facts, not a relationship the export states."""
+    got = jobs([{"JOBNAME": "JOB_X", "JOBCOUNT": "1",
+                 "AUTHCKNAM": "DDIC", "SDLUNAME": "JSMITH"}])
+    assert "JOBCMD-JOB-005" in got, got.keys()
+    rels = got["JOBCMD-JOB-005"]["relations"]
+    kinds = {(r["from"]["type"], r["to"]["type"]) for r in rels}
+    assert ("user", "job") in kinds and ("job", "user") in kinds
+    # ...and never the shortcut.
+    assert not [r for r in rels
+                if r["from"]["type"] == "user" and r["to"]["type"] == "user"], (
+        "a scheduler -> step user edge was declared: that is the transitive "
+        "claim the rules file refuses")
+
+
+def test_the_os_bridge_is_two_edges_from_the_job():
+    """The check's comment sketches "job -> program -> user", which reads as
+    though the program runs as the step user. It does not: the job runs the
+    program AND runs as the user."""
+    got = jobs([{"JOBNAME": "JOB_OS", "JOBCOUNT": "1",
+                 "PROGNAME": "RSBDCOS0", "AUTHCKNAM": "DDIC"}])
+    assert "JOBCMD-JOB-003" in got, got.keys()
+    rels = got["JOBCMD-JOB-003"]["relations"]
+    assert {(r["from"]["type"], r["to"]["type"]) for r in rels} == {
+        ("job", "program"), ("job", "user")}
+    assert not [r for r in rels if r["from"]["type"] == "program"], (
+        "an edge was declared FROM the program, which nothing in the export "
+        "supports")
+
+
 def test_every_rule_still_explains_itself():
     """`why` is what stops the rules file becoming a pile of type pairs nobody
     can audit."""