Test signing (#50)

* first draft (for testing only) to sign job payload

Signed-off-by: vsoch <[email protected]>

Author: vsoch

.devcontainer/Dockerfile

@@ -1,4 +1,4 @@
-FROM ghcr.io/rse-ops/accounting:app-latest
+FROM ghcr.io/rse-ops/pokemon:app-latest
 
 LABEL maintainer="Vanessasaurus <@vsoch>"
 
@@ -28,22 +28,9 @@ RUN python3 -m venv /env && \
 
 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}

.github/workflows/tests.yaml

@@ -48,4 +48,8 @@ jobs:
         export FLUX_REQUIRE_AUTH=true
         export TEST_AUTH=true
 
-        flux start pytest -xs tests/test_api.py
+        # This needs to be run as the flux instance owner
+        # E.g., if it's flux
+        # sudo -u flux flux start pytest -xs tests/test_api.py
+        whoami
+        sudo -u fluxuser flux start pytest -xs tests/test_api.py

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)
+ - Add better multi-user mode - running jobs on behalf of user (0.1.12)
  - Restore original rpc to get job info (has more information) (0.1.11)
  - Refactor of FLux Restful to use a database and OAauth2 (0.1.0)
  - Support for basic PAM authentication (0.0.11)

VERSION

@@ -1 +1 @@
-0.1.1
+0.1.12

app/core/security.py

@@ -13,21 +13,36 @@
 
 
 def create_access_token(
-    subject: Union[str, Any], expires_delta: timedelta = None
+    subject: Union[str, Any], expires_delta: timedelta = None, secret_key=None
 ) -> str:
+    """
+    Create a jwt access token.
+
+    We either use the user's secret key (which is hashed) or fall
+    back to the server set secret key.
+    """
+    # Use a user secret key, if they have one.
+    # Otherwise fall back to server secret key
+    secret_key = secret_key or settings.secret_key
     if expires_delta:
         expire = datetime.utcnow() + expires_delta
     else:
         expire = datetime.utcnow() + timedelta(
             minutes=settings.access_token_expires_minutes
         )
     to_encode = {"exp": expire, "sub": str(subject)}
-    return jwt.encode(to_encode, settings.secret_key, algorithm=ALGORITHM)
+    return jwt.encode(to_encode, secret_key, algorithm=ALGORITHM)
 
 
 def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """
+    Verify the password
+    """
     return pwd_context.verify(plain_password, hashed_password)
 
 
 def get_password_hash(password: str) -> str:
+    """
+    Note we aren't providing a salt here, so the same password can generate different.
+    """
     return pwd_context.hash(password)

app/crud/base.py

@@ -29,6 +29,8 @@ def get(self, db: Session, id: Any) -> Optional[ModelType]:
     def get_multi(
         self, db: Session, *, skip: int = 0, limit: int = 100
     ) -> List[ModelType]:
+        if limit is None:
+            return db.query(self.model).offset(skip).all()
         return db.query(self.model).offset(skip).limit(limit).all()
 
     def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:

app/library/flux.py

@@ -2,23 +2,67 @@
 import os
 import re
 import shlex
+import subprocess
 import time
 
 import flux
 import flux.job
 
 from app.core.config import settings
 
+root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+submit_script = os.path.join(root, "scripts", "submit-job.py")
 
-def submit_job(handle, jobspec, user):
+
+class FakeJob:
+    def __init__(self, jobid):
+        self.jobid = jobid
+
+    def get_id(self):
+        return self.jobid
+
+
+def submit_job(handle, fluxjob, user):
     """
-    Handle to submit a job, either with flux job submit or on behalf of user.
+    Submit the job on behalf of user.
     """
     if user and hasattr(user, "user_name"):
         print(f"User submitting job {user.user_name}")
+        user = user.user_name
     elif user and isinstance(user, str):
         print(f"User submitting job {user}")
-    return flux.job.submit_async(handle, jobspec)
+
+    # If we don't have auth enabled, submit in single-user mode
+    if not settings.require_auth:
+        print("Submit in single-user mode.")
+        return flux.job.submit_async(handle, fluxjob)
+
+    # Update the payload for the correct user
+    # Use helper script to sign payload
+    payload = json.dumps(fluxjob.jobspec)
+    # payload['HOME'] =
+
+    # We ideally need to pipe the payload into flux python
+    try:
+        ps = subprocess.Popen(("echo", payload), stdout=subprocess.PIPE)
+        output = subprocess.check_output(
+            ("sudo", "-E", "-u", user, "flux", "python", submit_script),
+            stdin=ps.stdout,
+            env=os.environ,
+        )
+        ps.wait()
+
+    # A flux start without sudo -u flux can cause this
+    # This will be caught and returned to the user
+    except PermissionError as e:
+        raise ValueError(
+            f"Permission error: {e}! Are you running the instance as the flux user?"
+        )
+
+    jobid = output.decode("utf-8").strip()
+    print("Submit job {jobid}")
+    job = FakeJob(jobid)
+    return job
 
 
 def validate_submit_kwargs(kwargs, envars=None, runtime=None):
@@ -98,8 +142,9 @@ def prepare_job(user, kwargs, runtime=0, workdir=None, envars=None):
     # Set an attribute about the owning user
     if user and hasattr(user, "user_name"):
         fluxjob.setattr("user", user.user_name)
-    elif isinstance(user, str):
-        fluxjob.setattr("user", user)
+        user = user.user_name
+
+    fluxjob.setattr("user", user)
 
     # Set a provided working directory
     print(f"⭐️ Workdir provided: {workdir}")

app/models/user.py

@@ -8,6 +8,10 @@ class User(Base):
     id = Column(Integer, primary_key=True, index=True)
     user_name = Column(String, unique=True, index=True, nullable=False)
     hashed_password = Column(String, nullable=False)
+
+    # These are only used to encode and validate payloads, and
+    # can be reset at any time. Not related to the hashed password
+    secret_key = Column(String, nullable=True)
     is_active = Column(Boolean(), default=True)
     is_superuser = Column(Boolean(), default=False)
     jobs = relationship("Job", back_populates="owner")

app/routers/api.py

@@ -190,9 +190,10 @@ async def submit_job(request: Request, user=user_auth):
     include everything in this function instead of having separate
     functions.
     """
-    print(f"User for submit is {user}")
     from app.main import app
 
+    print(f"User for submit is {user}")
+
     # This can bork if no payload is provided
     try:
         payload = await request.json()

app/routers/views.py

@@ -163,7 +163,6 @@ async def submit_job_post(request: Request, user=user_auth):
     Receive data posted (submit) to the form.
     """
     print(user)
-    from app.main import app
 
     messages = []
     form = SubmitForm(request)
@@ -177,7 +176,7 @@ async def submit_job_post(request: Request, user=user_auth):
                 launcher.launch(form.kwargs, workdir=form.workdir, user=user)
             )
         else:
-            return submit_job_helper(request, app, form, user=user)
+            return submit_job_helper(request, form, user=user)
     else:
         print("🍒 Submit form is NOT valid!")
     return templates.TemplateResponse(
@@ -192,10 +191,12 @@ async def submit_job_post(request: Request, user=user_auth):
     )
 
 
-def submit_job_helper(request, app, form, user):
+def submit_job_helper(request, form, user):
     """
     A helper to submit a flux job (not a launcher)
     """
+    from app.main import app
+
     # Submit the job and return the ID, but allow for error
     # Prepare the flux job! We don't support envars here yet
     try: