python: add require-instance job validator

Problem: It might be useful for a system instance to reject jobs which
are not themselves new instances of Flux, i.e. either batch jobs or
jobs that run `flux start` or `flux broker`. However, Flux does not
provide a way to do this.

Add a very simple `require-instance` job validator which attempts to
reject all jobs that do not create a new instance of Flux. It does
this by looking for either the "batch" system attribute in jobspec
(which will cause the job shell to start a new instance), or a command
that starts with "flux start" or "flux broker".

If necessary, this plugin could be copied and modified in environments
that want to be slightly more permissive, or allow other command
arguments that might result in a new instance. However, at least Flux
will provide an example from which to base new validators.

Fixes flux-framework#4214

src/bindings/python/flux/job/validator/plugins/require-instance.py

@@ -0,0 +1,39 @@
+##############################################################
+# Copyright 2022 Lawrence Livermore National Security, LLC
+# (c.f. AUTHORS, NOTICE.LLNS, COPYING)
+#
+# This file is part of the Flux resource manager framework.
+# For details, see https://github.com/flux-framework.
+#
+# SPDX-License-Identifier: LGPL-3.0
+##############################################################
+
+"""Require that all jobs are new instances of Flux
+
+This plugin validates that submitted jobs are new instances of Flux.
+That is, either the `batch` system attribute is set, or the first
+2 arguments of the command are `flux broker` or `flux start`.
+
+This is not a foolproof solution. The validator could reject jobs that
+are new instances of Flux if an instance is launched by a script and
+not directly via `flux broker` or `flux start`.
+"""
+
+import errno
+from os.path import basename
+from flux.job.validator import ValidatorPlugin
+
+
+class Validator(ValidatorPlugin):
+    def validate(self, args):
+        if "batch" in args.jobspec["attributes"]["system"]:
+            return
+        command = args.jobspec["tasks"][0]["command"]
+        arg0 = basename(command[0])
+        if arg0 == "flux" and command[1] in ["broker", "start"]:
+            return
+        return (
+            errno.EINVAL,
+            "Direct job submission is disabled for this instance."
+            + " Please use the batch or alloc subcommands of flux-mini(1)",
+        )