start of work to add flux python pam auth

.devcontainer/Dockerfile

@@ -1,10 +1,10 @@
-FROM fluxrm/flux-sched:focal
+FROM ghcr.io/rse-ops/accounting:app-latest
 
 LABEL maintainer="Vanessasaurus <@vsoch>"
 
 # Pip not provided in this version
 USER root
-RUN apt-get update && apt-get install -y python3-venv
+RUN apt-get update && apt-get install -y python3-venv systemctl
 COPY ./requirements.txt /requirements.txt
 COPY ./.github/dev-requirements.txt /dev-requirements.txt
 COPY ./docs/requirements.txt /docs-requirements.txt
@@ -18,11 +18,32 @@ RUN python3 -m pip install IPython && \
     python3 -m pip install -r /dev-requirements.txt && \
     python3 -m pip install -r /docs-requirements.txt
 
-    # Install isort and ensure on path
 RUN python3 -m venv /env && \
     . /env/bin/activate && \
     pip install -r /requirements.txt && \
     pip install -r /dev-requirements.txt && \
-    pip install -r /docs-requirements.txt
+    pip install -r /docs-requirements.txt && \
+    # Only for development - don't add this to a production container
+    sudo useradd -m -p $(openssl passwd '12345') "flux"
+
+RUN mkdir -p /run/flux /var/lib/flux mkdir /etc/flux/system/cron.d /mnt/curve && \
+    flux keygen /mnt/curve/curve.cert && \
+    # This probably needs to be done as flux user?
+    flux account create-db && \
+    flux account add-bank root 1 && \
+    flux account add-bank --parent-bank=root user_bank 1 && \
+    # These need to be owned by flux
+    chown -R flux /run/flux /var/lib/flux /mnt/curve && \
+    # flux-imp needs setuid permission
+    chmod u+s /usr/libexec/flux/flux-imp
+    # flux account add-user --username=fluxuser --bank=user_bank
+
+COPY ./example/multi-user/flux.service /etc/systemd/system/flux.service
+COPY ./example/multi-user/broker.toml /etc/flux/system/conf.d/broker.toml
+COPY ./example/multi-user/imp.toml /etc/flux/imp/conf.d/imp.toml
+
+RUN chmod 4755 /usr/libexec/flux/flux-imp \
+ && chmod 0644 /etc/flux/imp/conf.d/imp.toml \
+ && chmod 0644 /etc/flux/system/conf.d/broker.toml
 
 ENV PATH=/env/bin:${PATH}

.devcontainer/devcontainer.json

@@ -24,6 +24,6 @@
 		],
       },
     },
-    // Needed for git security feature
-    "postStartCommand": "git config --global --add safe.directory /workspaces/flux-python-api"
+    // Needed for git security feature, and flux config
+    "postStartCommand": "git config --global --add safe.directory /workspaces/flux-python-api && flux R encode --hosts=$(hostname) > /etc/flux/system/R && sed -i 's@HOSTNAME@'$(hostname)'@' /etc/flux/system/conf.d/broker.toml && sudo service munge start"
   }

.github/workflows/tests.yaml

@@ -29,6 +29,23 @@ jobs:
       run: pip install -r requirements.txt
     - name: Run tests
       run: |
+        # Tests for the API with auth disabled
         flux start pytest -xs tests/test_api.py
+
+        # Tests for the API with single user auth
+        export FLUX_REQUIRE_AUTH=true
         export TEST_AUTH=true
+        export FLUX_USER=fluxuser
+        export FLUX_TOKEN=12345
+        flux start pytest -xs tests/test_api.py
+
+        # Tests for the API with multi-user auth, but fail because user not created
+        unset FLUX_USER
+        unset FLUX_TOKEN
+        export TEST_PAM_AUTH=true
+        export TEST_PAM_AUTH_FAIL=true
+        export FLUX_ENABLE_PAM=true
         flux start pytest -xs tests/test_api.py
+
+        # TODO how to test pam in this mode?
+        # We would need to start flux as flux and run tests as a user

CHANGELOG.md

@@ -14,6 +14,7 @@ and **Merged pull requests**. Critical items to know are:
 The versions coincide with releases on pip. Only major versions will be released as tags on Github.
 
 ## [0.0.x](https://github.com/flux-framework/flux-restful-api/tree/main) (0.0.x)
+ - Support for basic PAM authentication (0.0.11)
  - Fixing bug with launcher always being specified (0.0.1)
   - catching any errors on creation of fluxjob
   - Add support uvicorn workers (>1 needed to run >1 process with Flux)

Dockerfile

@@ -11,6 +11,13 @@ ARG host="0.0.0.0"
 ARG workers="1"
 LABEL maintainer="Vanessasaurus <@vsoch>"
 
+ENV FLUX_USER=${user}
+ENV FLUX_TOKEN=${token}
+ENV FLUX_REQUIRE_AUTH=${use_auth}
+ENV PORT=${port}
+ENV HOST=${host}
+ENV WORKERS=${workers}
+
 USER root
 RUN apt-get update
 COPY ./requirements.txt /requirements.txt
@@ -27,10 +34,4 @@ RUN python3 -m pip install -r /requirements.txt && \
 
 WORKDIR /code
 COPY . /code
-ENV FLUX_USER=${user}
-ENV FLUX_TOKEN=${token}
-ENV FLUX_REQUIRE_AUTH=${use_auth}
-ENV PORT=${port}
-ENV HOST=${host}
-ENV WORKERS=${workers}
 ENTRYPOINT ["/code/entrypoint.sh"]

VERSION

@@ -1 +1 @@
-0.0.1
+0.0.11

app/core/config.py

@@ -67,6 +67,7 @@ class Settings(BaseSettings):
 
     # These map to envars, e.g., FLUX_USER
     has_gpus: bool = get_bool_envar("FLUX_HAS_GPUS")
+    enable_pam: bool = get_bool_envar("FLUX_ENABLE_PAM")
 
     # Assume there is at least one node!
     flux_nodes: int = get_int_envar("FLUX_NUMBER_NODES", 1)

app/library/auth.py

@@ -21,6 +21,7 @@ def not_authenticated(detail="Incorrect user or token."):
 
 def alert_auth():
     print("🍓 Require auth: %s" % settings.require_auth)
+    print("🍓     PAM auth: %s" % settings.enable_pam)
     print(
         "🍓    Flux user: %s" % ("*" * len(settings.flux_user))
         if settings.flux_user
@@ -33,25 +34,53 @@ def alert_auth():
     )
 
 
+def check_pam_auth(credentials: HTTPBasicCredentials = Depends(security)):
+    """
+    Check base64 encoded auth (this is HTTP Basic auth.)
+    """
+    # Ensure we have pam installed
+    try:
+        import pam
+    except ImportError:
+        print("python-pam is required for PAM.")
+        return
+
+    username = credentials.username.encode("utf8")
+    password = credentials.password.encode("utf8")
+    if pam.authenticate(username, password) is True:
+        return credentials.username
+
+
 def check_auth(credentials: HTTPBasicCredentials = Depends(security)):
     """
     Check base64 encoded auth (this is HTTP Basic auth.)
     """
+    # First try to authenticate with PAM, if allowed.
+    if settings.enable_pam:
+        print("🧾️ Checking PAM auth...")
+        # Return the username if PAM authentication is successful
+        username = check_pam_auth(credentials)
+        if username:
+            print("🧾️ Success!")
+            return username
+
+    # If we get here, we require the flux user and token
     if not settings.flux_user or not settings.flux_token:
-        return not_authenticated("Missing FLUX_USER and/or FLUX_TOKEN")
+        return not_authenticated("Missing FLUX_USER and/or FLUX_TOKEN or pam headers")
+
     current_username_bytes = credentials.username.encode("utf8")
     correct_username_bytes = bytes(settings.flux_user.encode("utf8"))
     is_correct_username = secrets.compare_digest(
         current_username_bytes, correct_username_bytes
     )
-    current_password_bytes = credentials.password.encode("utf8")
 
+    current_password_bytes = credentials.password.encode("utf8")
     correct_password_bytes = bytes(settings.flux_token.encode("utf8"))
     is_correct_password = secrets.compare_digest(
         current_password_bytes, correct_password_bytes
     )
     if not (is_correct_username and is_correct_password):
-        return not_authenticated()
+        return not_authenticated("heree")
     return credentials.username
 
 

app/library/flux.py

@@ -1,14 +1,41 @@
 import json
 import os
+import pwd
 import re
 import shlex
 import time
 
 import flux
 import flux.job
 
+import app.library.terminal as terminal
 from app.core.config import settings
 
+# Faux user environment (filtered set of application environment)
+# We could likely find a way to better do this, but likely the users won't have customized environments
+user_env = {
+    "SHELL": "/bin/bash",
+    "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/snap/bin",
+    "XDG_RUNTIME_DIR": "/tmp/user/0",
+    "DISPLAY": ":0",
+    "COLORTERM": "truecolor",
+    "SHLVL": "2",
+    "DEBIAN_FRONTEND": "noninteractive",
+    "MAKE_TERMERR": "/dev/pts/1",
+    "LANG": "C.UTF-8",
+    "TERM": "xterm-256color",
+}
+
+
+def submit_job(handle, jobspec, user):
+    """
+    Handle to submit a job, either with flux job submit or on behalf of user.
+    """
+    # We've enabled PAM auth
+    if settings.enable_pam:
+        return terminal.submit_job(jobspec, user)
+    return flux.job.submit_async(handle, jobspec)
+
 
 def validate_submit_kwargs(kwargs, envars=None, runtime=None):
     """
@@ -68,6 +95,7 @@ def prepare_job(kwargs, runtime=0, workdir=None, envars=None):
     command = kwargs["command"]
     if isinstance(command, str):
         command = shlex.split(command)
+
     print(f"⭐️ Command being submit: {command}")
 
     # Delete command from the kwargs (we added because is required and validated that way)
@@ -90,8 +118,14 @@ def prepare_job(kwargs, runtime=0, workdir=None, envars=None):
     # A duration of zero (the default) means unlimited
     fluxjob.duration = runtime
 
+    # If we are running as the user, we don't want the current (root) environment
+    # This isn't perfect because it's artifically created, but it ensures we have paths
+    if settings.enable_pam:
+        environment = user_env
+    else:
+        environment = dict(os.environ)
+
     # Additional envars in the payload?
-    environment = dict(os.environ)
     environment.update(envars)
     fluxjob.environment = environment
     return fluxjob
@@ -131,12 +165,15 @@ def stream_job_output(jobid):
         pass
 
 
-def cancel_job(jobid):
+def cancel_job(jobid, user):
     """
     Request a job to be cancelled by id.
 
     Returns a message to the user and a return code.
     """
+    if settings.enable_pam:
+        return terminal.cancel_job(jobid, user)
+
     from app.main import app
 
     try:
@@ -147,12 +184,16 @@ def cancel_job(jobid):
     return "Job is requested to cancel.", 200
 
 
-def get_job_output(jobid, delay=None):
+def get_job_output(jobid, user=None, delay=None):
     """
     Given a jobid, get the output.
 
     If there is a delay, we are requesting on demand, so we want to return early.
     """
+    # We've enabled PAM auth
+    if settings.enable_pam:
+        return terminal.get_job_output(jobid, user, delay=delay)
+
     lines = []
     start = time.time()
     from app.main import app
@@ -171,38 +212,48 @@ def get_job_output(jobid, delay=None):
     return lines
 
 
-def list_jobs_detailed(limit=None, query=None):
+def list_jobs_detailed(user=None, limit=None, query=None):
     """
     Get a detailed listing of jobs.
     """
-    listing = list_jobs()
+    listing = list_jobs(user=user)
     ids = listing.get()["jobs"]
     jobs = {}
     for job in ids:
-
         # Stop if a limit is defined and we have hit it!
         if limit is not None and len(jobs) >= limit:
             break
 
         try:
-            jobinfo = get_job(job["id"])
+            jobinfo = get_job(job["id"], user=user)
 
             # Best effort hack to do a query
             if query and not query_job(jobinfo, query):
                 continue
+
+            # This will trigger a data table warning
+            for needed in ["ranks", "expiration"]:
+                if needed not in jobinfo:
+                    jobinfo[needed] = ""
+
             jobs[job["id"]] = jobinfo
+
         except Exception:
             pass
     return jobs
 
 
-def list_jobs():
+def list_jobs(user=None):
     """
     Get a simple listing of jobs (just the ids)
     """
     from app.main import app
 
-    return flux.job.job_list(app.handle)
+    if user is None or not settings.enable_pam:
+        return flux.job.job_list(app.handle)
+    pw_record = pwd.getpwnam(user)
+    user_uid = pw_record.pw_uid
+    return flux.job.job_list(app.handle, userid=user_uid)
 
 
 def get_simple_job(jobid):
@@ -215,13 +266,17 @@ def get_simple_job(jobid):
     return json.loads(info.get_str())["job"]
 
 
-def get_job(jobid):
+def get_job(jobid, user):
     """
     Get details for a job
     """
     from app.main import app
 
-    payload = {"id": int(jobid), "attrs": ["all"]}
+    jobid = flux.job.JobID(jobid)
+
+    payload = {"id": jobid, "attrs": ["all"]}
+    if settings.enable_pam:
+        payload["user"] = user
     rpc = flux.job.list.JobListIdRPC(app.handle, "job-list.list-id", payload)
     try:
         jobinfo = rpc.get()