sftpgo: configure for credential-gated r/w access

hosts/by-name/servo/services/export/sftpgo.nix

@@ -69,11 +69,14 @@ in
         };
 
         banner = ''
-          Welcome, friends, to Colin's read-only FTP server! Also available via NFS on the same host.
+          Welcome, friends, to Colin's FTP server! Also available via NFS on the same host, but LAN-only.
+
+          Read-only access:
           Username: "anonymous"
           Password: "anonymous"
+
           CONFIGURE YOUR CLIENT FOR "PASSIVE" mode, e.g. `ftp --passive uninsane.org`
-          Please let me know if anything's broken or not as it should be. Otherwise, browse and DL freely :)
+          Please let me know if anything's broken or not as it should be. Otherwise, browse and transfer freely :)
         '';
 
       };

hosts/by-name/servo/services/export/sftpgo_external_auth_hook

@@ -37,9 +37,12 @@
 # - it seems (empirically) that a user can't cd above their home directory.
 #   though i don't have a reference for that in the docs.
 
+import crypt
 import json
 import os
 
+from hmac import compare_digest
+
 authFail = dict(username="")
 
 PERM_RO = [ "list", "download" ]
@@ -60,6 +63,13 @@ PERM_RW = [
   # "chtimes",
 ]
 
+TRUSTED_CREDS = [
+  # /etc/shadow style creds.
+  # mkpasswd -m sha-512
+  # $<method>$<salt>$<hash>
+  "$6$Zq3c2u4ghUH4S6EP$pOuRt13sEKfX31OqPbbd1LuhS21C9MICMc94iRdTAgdAcJ9h95gQH/6Jf6Ie4Obb0oxQtojRJ1Pd/9QHOlFMW."  #< m. rocket boy
+]
+
 def mkAuthOk(username: str, permissions: dict[str, list[str]]) -> dict:
   return dict(
     status = 1,
@@ -100,11 +110,32 @@ def isLan(ip: str) -> bool:
 def isWireguard(ip: str) -> bool:
   return ip.startswith("10.0.10.")
 
-def getAuthResponse(ip: str, username: str) -> dict:
+def isTrustedCred(password: str) -> bool:
+  for cred in TRUSTED_CREDS:
+    _, method, salt, hash_ = cred.split("$")
+    # assert method == "6", f"unrecognized crypt entry: {cred}"
+    if crypt.crypt(password, f"${method}${salt}") == cred:
+        return True
+
+  return False
+
+def getAuthResponse(ip: str, username: str, password: str) -> dict:
   """
   return a sftpgo auth response either denying the user or approving them
   with a set of permissions.
   """
+  if isTrustedCred(password) and username != "colin":
+    # allow r/w access from those with a special token
+    return mkAuthOk(username, permissions = {
+      "/": PERM_RW,
+      "/playground": PERM_RW,
+    })
+  if isWireguard(ip):
+    # allow any user from wireguard
+    return mkAuthOk(username, permissions = {
+      "/": PERM_RW,
+      "/playground": PERM_RW,
+    })
   if isLan(ip):
     if username == "anonymous":
       # allow anonymous users on the LAN
@@ -112,19 +143,14 @@ def getAuthResponse(ip: str, username: str) -> dict:
         "/": PERM_RO,
         "/playground": PERM_RW,
       })
-  if isWireguard(ip):
-    # allow any user from wireguard
-    return mkAuthOk(username, permissions = {
-      "/": PERM_RW,
-      "/playground": PERM_RW,
-    })
 
   return authFail
 
 def main():
   ip = os.environ.get("SFTPGO_AUTHD_IP", "")
   username = os.environ.get("SFTPGO_AUTHD_USERNAME", "")
-  resp = getAuthResponse(ip, username)
+  password = os.environ.get("SFTPGO_AUTHD_PASSWORD", "")
+  resp = getAuthResponse(ip, username, password)
   print(json.dumps(resp))
 
 if __name__ == "__main__":