bicep/matrix: enable smtp auth

yolo lmao

hosts/bicep/services/matrix/smtp-authenticator/default.nix

@@ -0,0 +1,17 @@
+{ lib, buildPythonPackage, fetchFromGitHub }:
+
+buildPythonPackage rec {
+  pname = "matrix-synapse-smtp-auth";
+  version = "0.1.0";
+
+  src = ./.;
+
+  doCheck = false;
+
+  meta = with lib; {
+    description = "An SMTP auth provider for Synapse";
+    homepage = "pvv.ntnu.no";
+    license = licenses.agpl3Only;
+    maintainers = with maintainers; [ dandellion ];
+  };
+}

hosts/bicep/services/matrix/smtp-authenticator/setup.py

@@ -0,0 +1,11 @@
+from setuptools import setup
+
+setup(
+    name="matrix-synapse-smtp-auth",
+    version="0.1.0",
+    py_modules=['smtp_auth_provider'],
+    author="Daniel Løvbrøtte Olsen",
+    author_email="danio@pvv.ntnu.no",
+    description="An SMTP auth provider for Synapse",
+    license="AGPL-3.0-only"
+)

hosts/bicep/services/matrix/smtp-authenticator/smtp_auth_provider.py

@@ -0,0 +1,45 @@
+from typing import Awaitable, Callable, Optional, Tuple
+
+from smtplib import SMTP_SSL as SMTP
+
+import synapse
+from synapse import module_api
+
+
+class SMTPAuthProvider:
+    def __init__(self, config: dict, api: module_api):
+        self.api = api
+
+        self.config = config
+
+        api.register_password_auth_provider_callbacks(
+            auth_checkers={
+                ("m.login.password", ("password",)): self.check_pass,
+            },
+        )
+
+    async def check_pass(
+        self,
+        username: str,
+        login_type: str,
+        login_dict: "synapse.module_api.JsonDict",
+    ):
+        if login_type != "m.login.password":
+            return None
+
+        result = False
+        with SMTP(self.config["smtp_host"]) as smtp:
+            password = login_dict.get("password")
+            try:
+                smtp.login(username, password)
+                result = True
+            except:
+                return None
+
+        if result == True:
+            userid = self.api.get_qualified_user_id(username)
+            if not self.api.check_user_exists(userid):
+                self.api.register_user(username)
+            return (userid, None)
+        else:
+            return None