feat: basic protection boot endpoint (#334)

* feat: basic protection boot endpoint

* rename env variable to indicate experimental status

* update with more test cases

* add command line flags for boot protection

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,8 @@
 
 
 def setup(name, bran, adminPort, bootPort, base='', httpPort=None, configFile=None, configDir=None,
-          keypath=None, certpath=None, cafilepath=None, cors=False, releaseTimeout=None, curls=None, iurls=None, durls=None):
+          keypath=None, certpath=None, cafilepath=None, cors=False, releaseTimeout=None, curls=None,
+          iurls=None, durls=None, bootUsername=None, bootPassword=None):
     """ Set up an ahab in Signify mode """
 
     agency = Agency(name=name, base=base, bran=bran, configFile=configFile, configDir=configDir, releaseTimeout=releaseTimeout, curls=curls, iurls=iurls, durls=durls)
@@ -66,7 +68,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())
 
@@ -871,17 +873,44 @@ 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):
+        # Username AND Password is not set, so no need to authenticate
+        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 is None or password is None:
+                raise falcon.HTTPUnauthorized(title="Unauthorized")
+
+            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
@@ -893,6 +922,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

@@ -68,6 +68,14 @@
                     help="Set log level to DEBUG | INFO | WARNING | ERROR | CRITICAL. Default is CRITICAL")
 parser.add_argument("--logfile", action="store", required=False, default=None,
                     help="path of the log file. If not defined, logs will not be written to the file.")
+parser.add_argument("--experimental-boot-password",
+                    help="Experimental password for boot endpoint. Enables HTTP Basic Authentication for the boot endpoint. Only meant to be used for testing purposes.",
+                    dest="bootPassword",
+                    default=os.getenv("KERIA_EXPERIMENTAL_BOOT_PASSWORD"))
+parser.add_argument("--experimental-boot-username",
+                    help="Experimental username for boot endpoint. Enables HTTP Basic Authentication for the boot endpoint. Only meant to be used for testing purposes.",
+                    dest="bootUsername",
+                    default=os.getenv("KERIA_EXPERIMENTAL_BOOT_USERNAME"))
 
 def getListVariable(name):
     value = os.getenv(name)
@@ -99,7 +107,9 @@ def launch(args):
                             releaseTimeout=int(os.getenv("KERIA_RELEASER_TIMEOUT", "86400")),
                             curls=getListVariable("KERIA_CURLS"),
                             iurls=getListVariable("KERIA_IURLS"),
-                            durls=getListVariable("KERIA_DURLS"))
+                            durls=getListVariable("KERIA_DURLS"),
+                            bootPassword=args.bootPassword,
+                            bootUsername=args.bootUsername)
 
     directing.runController(doers=agency, expire=0.0)
 

tests/app/test_agenting.py

@@ -5,6 +5,7 @@
 
 Testing the Mark II Agent
 """
+from base64 import b64encode
 import json
 import os
 import shutil
@@ -248,7 +249,7 @@ def test_agency_with_urls_from_arguments():
         assert agent.hby.cf.get()["iurls"] == iurls
         assert agent.hby.cf.get()["durls"] == durls
 
-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])
@@ -281,6 +282,85 @@ 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 = "user"
+    password = "secret"
+
+    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 user:secret"})
+    assert rep.status_code == 401
+
+    rep = client.simulate_post("/boot", body=json.dumps(body).encode("utf-8"), headers={"Authorization": f"Basic {b64encode(b'test:secret').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'user').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'user:test').decode('utf-8')}"} )
+    assert rep.status_code == 401
+
+    authorization = f"Basic {b64encode(b'user:secret').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_misconfigured_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)
+
+    # No password set, should return 401
+    bootEnd = agenting.BootEnd(agency, username="user", password=None)
+    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]
+        )
+    )
+
+    authorization = f"Basic {b64encode(b'user').decode('utf-8')}"
+    rep = client.simulate_post("/boot", body=json.dumps(body).encode("utf-8"), headers={"Authorization": authorization})
+    assert rep.status_code == 401
+
+    authorization = f"Basic {b64encode(b'user:secret').decode('utf-8')}"
+    rep = client.simulate_post("/boot", body=json.dumps(body).encode("utf-8"), headers={"Authorization": authorization})
+    assert rep.status_code == 401
 
 def test_witnesser(helpers):
     salt = b'0123456789abcdef'