feat: basic protection boot endpoint

src/keria/app/agenting.py

@@ -4,6 +4,7 @@
 keria.app.agenting module
 
 """
+from base64 import b64decode
 import json
 import os
 import datetime
@@ -53,7 +54,7 @@
 
 
 def setup(name, bran, adminPort, bootPort, base='', httpPort=None, configFile=None, configDir=None,
-          keypath=None, certpath=None, cafilepath=None):
+          keypath=None, certpath=None, cafilepath=None, bootUsername=None, bootPassword=None):
     """ Set up an ahab in Signify mode """
 
     agency = Agency(name=name, base=base, bran=bran, configFile=configFile, configDir=configDir)
@@ -66,7 +67,7 @@ def setup(name, bran, adminPort, bootPort, base='', httpPort=None, configFile=No
     if not bootServer.reopen():
         raise RuntimeError(f"cannot create boot http server on port {bootPort}")
     bootServerDoer = http.ServerDoer(server=bootServer)
-    bootEnd = BootEnd(agency)
+    bootEnd = BootEnd(agency, username=bootUsername, password=bootPassword)
     bootApp.add_route("/boot", bootEnd)
     bootApp.add_route("/health", HealthEnd())
 
@@ -857,17 +858,40 @@ def loadEnds(app):
 class BootEnd:
     """ Resource class for creating datastore in cloud ahab """
 
-    def __init__(self, agency):
+    def __init__(self, agency: Agency, username: str | None = None, password: str | None = None):
         """ Provides endpoints for initializing and unlocking an agent
-
         Parameters:
             agency (Agency): Agency for managing agents
-
+            username (str): username for boot request
+            password (str): password for boot request
         """
-        self.authn = authing.Authenticater(agency=agency)
+        self.username = username
+        self.password = password
         self.agency = agency
 
-    def on_post(self, req, rep):
+    def authenticate(self, req: falcon.Request):
+        if self.username is None and self.password is None:
+            return
+
+        if req.auth is None:
+            raise falcon.HTTPUnauthorized(title="Unauthorized")
+
+        scheme, token = req.auth.split(' ')
+        if scheme != 'Basic':
+            raise falcon.HTTPUnauthorized(title="Unauthorized")
+
+        try:
+            username, password = b64decode(token).decode('utf-8').split(':')
+
+            if username == self.username and password == self.password:
+                return
+
+        except Exception:
+            raise falcon.HTTPUnauthorized(title="Unauthorized")
+
+        raise falcon.HTTPUnauthorized(title="Unauthorized")
+
+    def on_post(self, req: falcon.Request, rep: falcon.Response):
         """ Inception event POST endpoint
 
         Give me a new Agent.  Create Habery using ctrlPRE as database name, agentHab that anchors the caid and
@@ -879,6 +903,8 @@ def on_post(self, req, rep):
 
         """
 
+        self.authenticate(req)
+
         body = req.get_media()
         if "icp" not in body:
             raise falcon.HTTPBadRequest(title="invalid inception",

src/keria/app/cli/commands/start.py

@@ -7,6 +7,7 @@
 """
 import argparse
 import logging
+import os
 
 from keri import __version__
 from keri import help
@@ -108,6 +109,8 @@ def runAgent(name="ahab", base="", bran="", admin=3901, http=3902, boot=3903, co
                                 configDir=configDir,
                                 keypath=keypath,
                                 certpath=certpath,
-                                cafilepath=cafilepath))
+                                cafilepath=cafilepath,
+                                bootPassword=os.getenv("KERIA_BOOT_PASSWORD"),
+                                bootUsername=os.getenv("KERIA_BOOT_USERNAME")))
 
     directing.runController(doers=doers, expire=expire)

tests/app/test_agenting.py

@@ -5,6 +5,7 @@
 
 Testing the Mark II Agent
 """
+from base64 import b64encode
 import json
 import os
 import shutil
@@ -202,7 +203,7 @@ def test_agency():
         assert caid not in agency.agents
         assert len(agent.doers) == 0
 
-def test_boot_ends(helpers):
+def test_unprotected_boot_ends(helpers):
     agency = agenting.Agency(name="agency", bran=None, temp=True)
     doist = doing.Doist(limit=1.0, tock=0.03125, real=True)
     doist.enter(doers=[agency])
@@ -235,6 +236,51 @@ def test_boot_ends(helpers):
         'description': 'agent for controller EK35JRNdfVkO4JwhXaSTdV4qzB_ibk_tGJmSVcY4pZqx already exists'
     }
 
+def test_protected_boot_ends(helpers):
+    agency = agenting.Agency(name="agency", bran=None, temp=True)
+    doist = doing.Doist(limit=1.0, tock=0.03125, real=True)
+    doist.enter(doers=[agency])
+
+    serder, sigers = helpers.controller()
+    assert serder.pre == helpers.controllerAID
+
+    app = falcon.App()
+    client = testing.TestClient(app)
+
+    username = "test"
+    password = "test"
+
+    bootEnd = agenting.BootEnd(agency, username=username, password=password)
+    app.add_route("/boot", bootEnd)
+
+    body = dict(
+        icp=serder.ked,
+        sig=sigers[0].qb64,
+        salty=dict(
+            stem='signify:aid', pidx=0, tier='low', sxlt='OBXYZ',
+            icodes=[MtrDex.Ed25519_Seed], ncodes=[MtrDex.Ed25519_Seed]
+        )
+    )
+
+    rep = client.simulate_post("/boot", body=json.dumps(body).encode("utf-8"))
+    assert rep.status_code == 401
+
+    rep = client.simulate_post("/boot", body=json.dumps(body).encode("utf-8"), headers={"Authorization": "Something test"})
+    assert rep.status_code == 401
+
+    rep = client.simulate_post("/boot", body=json.dumps(body).encode("utf-8"), headers={"Authorization": "Basic test:test"})
+    assert rep.status_code == 401
+
+    rep = client.simulate_post("/boot", body=json.dumps(body).encode("utf-8"), headers={"Authorization": f"Basic {b64encode(b'test:foobar').decode('utf-8')}"} )
+    assert rep.status_code == 401
+
+    rep = client.simulate_post("/boot", body=json.dumps(body).encode("utf-8"), headers={"Authorization": f"Basic {b64encode(b'foobar:test').decode('utf-8')}"} )
+    assert rep.status_code == 401
+
+    authorization = f"Basic {b64encode(b'test:test').decode('utf-8')}"
+    rep = client.simulate_post("/boot", body=json.dumps(body).encode("utf-8"), headers={"Authorization": authorization})
+    assert rep.status_code == 202
+
 
 def test_witnesser(helpers):
     salt = b'0123456789abcdef'