Initial Awawawa

web/tests/helpers.py

@@ -0,0 +1,55 @@
+import os
+import re
+import shutil
+import tempfile
+import unittest
+from unittest import mock
+
+from patchbay_web import create_app
+
+ED25519 = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAINJ2M1DCOTYJkGHaPO2iMGqF7isZXmwqN5HxtLP5+xq1 test@host"
+
+
+class AppCase(unittest.TestCase):
+    """Fresh config + database per test; mails are captured instead of sent."""
+
+    def setUp(self):
+        self.dir = tempfile.mkdtemp(prefix="pbtest.")
+        self.conf_path = os.path.join(self.dir, "patchbay.conf")
+        with open(self.conf_path, "w") as f:
+            f.write("# test config\nRole = target\n"
+                    f"Database = {self.dir}/patchbay.db\nRunDir = {self.dir}\n"
+                    "SysopUser = admin\nSysopEmail = admin@example.org\n"
+                    "SysopPassword = correct horse battery\nMailHost = smtp.invalid\n")
+        self.mails = []
+        patcher = mock.patch("patchbay_web.mailer.send",
+                             side_effect=lambda conf, to, subj, body: self.mails.append((to, body)))
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        self.app = create_app(self.conf_path, testing=True)
+        self.client = self.app.test_client()
+
+    def tearDown(self):
+        shutil.rmtree(self.dir)
+
+    def csrf(self):
+        page = self.client.get("/login").get_data(as_text=True)
+        m = re.search(r'name="csrf-token" content="([^"]+)"', page)
+        return m.group(1)
+
+    def login(self, user="admin", password="correct horse battery"):
+        tok = self.csrf()
+        r = self.client.post("/login", data={"username": user, "password": password, "csrf_token": tok})
+        if r.status_code != 302:
+            return r
+        code = re.search(r"code is: (\d{6})", self.mails[-1][1]).group(1)
+        page = self.client.get("/verify").get_data(as_text=True)
+        tok = re.search(r'name="csrf-token" content="([^"]+)"', page).group(1)
+        return self.client.post("/verify", data={"code": code, "csrf_token": tok})
+
+    def token_from(self, path):
+        page = self.client.get(path).get_data(as_text=True)
+        return re.search(r'name="csrf-token" content="([^"]+)"', page).group(1)
+
+    def token(self):
+        return self.token_from("/settings")

web/tests/test_auth.py

@@ -0,0 +1,75 @@
+import unittest
+
+from patchbay_web import security
+from patchbay_web.conf import Config
+
+from .helpers import AppCase
+
+
+class AuthTest(AppCase):
+    def test_sysop_password_hashed_on_start(self):
+        conf = Config(self.conf_path)
+        self.assertTrue(conf.get("sysoppassword").startswith("$scrypt$"))
+        with open(self.conf_path) as f:
+            self.assertIn("# test config", f.read())
+
+    def test_login_2fa(self):
+        r = self.login()
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(self.mails[-1][0], "admin@example.org")
+        self.assertEqual(self.client.get("/patch").status_code, 200)
+
+    def test_wrong_password(self):
+        r = self.login(password="nope")
+        self.assertEqual(r.status_code, 401)
+        self.assertEqual(self.mails, [])
+
+    def test_wrong_code_then_expired_attempts(self):
+        tok = self.csrf()
+        self.client.post("/login", data={"username": "admin", "password": "correct horse battery", "csrf_token": tok})
+        tok = self.token_from("/verify")
+        for _ in range(5):
+            r = self.client.post("/verify", data={"code": "000000", "csrf_token": tok})
+        r = self.client.post("/verify", data={"code": "000000", "csrf_token": tok})
+        self.assertEqual(r.status_code, 302)  # sent back to login
+        self.assertEqual(self.client.get("/patch").status_code, 302)
+
+    def test_csrf_required(self):
+        self.login()
+        r = self.client.post("/settings/users/add", data={"username": "bob"})
+        self.assertEqual(r.status_code, 400)
+
+    def test_requires_login(self):
+        self.assertEqual(self.client.get("/settings").status_code, 302)
+        self.assertEqual(self.client.get("/api/graph").status_code, 401)
+
+    def test_user_management_sysop_only(self):
+        self.login()
+        tok = self.token()
+        r = self.client.post("/settings/users/add", data={
+            "username": "bob", "email": "bob@example.org", "password": "0123456789x", "csrf_token": tok})
+        self.assertEqual(r.status_code, 302)
+        self.client.post("/logout", data={"csrf_token": tok})
+
+        self.assertEqual(self.login("bob", "0123456789x").status_code, 302)
+        tok = self.token()
+        r = self.client.post("/settings/users/add", data={
+            "username": "eve", "email": "e@example.org", "password": "0123456789x", "csrf_token": tok})
+        self.assertEqual(r.status_code, 403)
+
+    def test_rate_limit(self):
+        for _ in range(10):
+            self.login(password="bad")
+        self.assertEqual(self.login().status_code, 429)
+
+
+class SecurityTest(unittest.TestCase):
+    def test_hash_roundtrip(self):
+        h = security.hash_password("hunter22")
+        self.assertTrue(security.verify_password("hunter22", h))
+        self.assertFalse(security.verify_password("hunter23", h))
+        self.assertFalse(security.verify_password("x", None))
+
+
+if __name__ == "__main__":
+    unittest.main()

web/tests/test_conf.py

@@ -0,0 +1,27 @@
+import os
+import tempfile
+import unittest
+
+from patchbay_web.conf import Config
+
+
+class ConfTest(unittest.TestCase):
+    def test_update_preserves_layout(self):
+        with tempfile.TemporaryDirectory() as d:
+            p = os.path.join(d, "patchbay.conf")
+            with open(p, "w") as f:
+                f.write("# head\nRole = target\nmailhost = old\nMailHost = dup\n")
+            os.chmod(p, 0o600)
+            c = Config(p)
+            self.assertEqual(c.get("MailHost"), "dup")
+            c.update({"MailHost": "smtp.example.org", "MailPort": 465})
+            with open(p) as f:
+                text = f.read()
+            self.assertEqual(text, "# head\nRole = target\nmailhost = smtp.example.org\nMailPort = 465\n")
+            self.assertEqual(os.stat(p).st_mode & 0o777, 0o600)
+            with self.assertRaises(ValueError):
+                c.update({"MailHost": "a\nRole = client"})
+
+
+if __name__ == "__main__":
+    unittest.main()

web/tests/test_graph.py

@@ -0,0 +1,96 @@
+import unittest
+
+from patchbay_web import validate
+
+from .helpers import ED25519, AppCase
+
+
+class ValidateTest(unittest.TestCase):
+    def test_pubkey(self):
+        self.assertEqual(validate.pubkey(ED25519), " ".join(ED25519.split()[:2]))
+        for bad in ["", "ssh-ed25519", "ssh-dss AAAA", "ssh-ed25519 !!!!",
+                    'ssh-rsa AAAAC3NzaC1lZDI1NTE5AAAAINJ2M1DCOTYJkGHaPO2iMGqF7isZXmwqN5HxtLP5+xq1']:
+            with self.assertRaises(validate.Invalid):
+                validate.pubkey(bad)
+
+    def test_graph_rules(self):
+        src = {"id": -1, "type": "client_source", "client_id": 1, "port": 22, "proto": "tcp"}
+        sink = {"id": -2, "type": "public_sink", "host": "0.0.0.0", "port": 2200, "proto": "tcp"}
+        split = {"id": -3, "type": "splitter"}
+        nodes, links = validate.graph({"nodes": [src, sink], "links": [{"from": -1, "to": -2}]}, {1})
+        self.assertEqual(links, [(-1, -2)])
+        with self.assertRaises(validate.Invalid):  # wrong direction
+            validate.graph({"nodes": [src, sink], "links": [{"from": -2, "to": -1}]}, {1})
+        with self.assertRaises(validate.Invalid):  # unknown client
+            validate.graph({"nodes": [src], "links": []}, set())
+        with self.assertRaises(validate.Invalid):  # source with two outputs
+            sink2 = dict(sink, id=-4)
+            validate.graph({"nodes": [src, sink, sink2],
+                            "links": [{"from": -1, "to": -2}, {"from": -1, "to": -4}]}, {1})
+        with self.assertRaises(validate.Invalid):  # loop through splitters
+            split2 = dict(split, id=-5)
+            validate.graph({"nodes": [split, split2],
+                            "links": [{"from": -3, "to": -5}, {"from": -5, "to": -3}]}, {1})
+        with self.assertRaises(validate.Invalid):
+            validate.graph({"nodes": [dict(sink, host="a b")], "links": []}, {1})
+
+    def test_tunnel_nodes(self):
+        tsrc = {"id": -1, "type": "tunnel_source", "client_id": None, "host": "10.100.0.2", "port": 25565,
+                "proto": "tcp", "iface": "tun0"}
+        tsink = {"id": -2, "type": "tunnel_sink", "client_id": 1, "port": 8080, "proto": "tcp", "iface": "tun1"}
+        nodes, links = validate.graph({"nodes": [tsrc, tsink], "links": [{"from": -1, "to": -2}]}, {1})
+        self.assertEqual(nodes[0]["iface"], "tun0")
+        self.assertIsNone(nodes[0]["client_id"])
+        self.assertEqual(links, [(-1, -2)])
+        with self.assertRaises(validate.Invalid):
+            validate.graph({"nodes": [dict(tsrc, iface="tun0; rm")], "links": []}, {1})
+        # iface is ignored on non-tunnel nodes
+        n, _ = validate.graph({"nodes": [{"id": -3, "type": "public_sink", "port": 1, "iface": "x"}], "links": []}, {1})
+        self.assertEqual(n[0]["iface"], "")
+
+
+class GraphApiTest(AppCase):
+    def test_roundtrip_keeps_ids(self):
+        self.login()
+        tok = self.token()
+        r = self.client.post("/settings/clients/add", data={"name": "c1", "pubkey": ED25519, "csrf_token": tok})
+        self.assertEqual(r.status_code, 302)
+        cid = self.client.get("/api/graph").get_json()["clients"][0]["id"]
+
+        payload = {
+            "nodes": [
+                {"id": -1, "type": "client_source", "client_id": cid, "host": "127.0.0.1", "port": 22, "proto": "tcp", "x": 0, "y": 0},
+                {"id": -2, "type": "splitter", "x": 100, "y": 0},
+                {"id": -3, "type": "public_sink", "host": "0.0.0.0", "port": 2200, "proto": "tcp", "x": 200, "y": 0},
+            ],
+            "links": [{"from": -1, "to": -2}, {"from": -2, "to": -3}],
+        }
+        r = self.client.put("/api/graph", json=payload, headers={"X-CSRF-Token": tok})
+        self.assertEqual(r.status_code, 200, r.get_data(as_text=True))
+        ids = r.get_json()["ids"]
+        self.assertFalse(r.get_json()["daemon"])  # no daemon in tests
+
+        g = self.client.get("/api/graph").get_json()
+        self.assertEqual(len(g["nodes"]), 3)
+        sink_id = ids["-3"]
+        # Saving again with real ids keeps them (stats are keyed by sink id).
+        for n in payload["nodes"]:
+            n["id"] = ids[str(n["id"])]
+        payload["links"] = [{"from": ids["-1"], "to": ids["-2"]}, {"from": ids["-2"], "to": ids["-3"]}]
+        r = self.client.put("/api/graph", json=payload, headers={"X-CSRF-Token": tok})
+        self.assertEqual(r.get_json()["ids"][str(sink_id)], sink_id)
+
+        # Deleting the client keeps its nodes but clears the reference.
+        self.client.post(f"/settings/clients/{cid}/delete", data={"csrf_token": tok})
+        g = self.client.get("/api/graph").get_json()
+        self.assertIsNone(next(n for n in g["nodes"] if n["type"] == "client_source")["client_id"])
+
+    def test_invalid_graph_rejected(self):
+        self.login()
+        tok = self.token()
+        r = self.client.put("/api/graph", json={"nodes": [{"id": -1, "type": "bogus"}]}, headers={"X-CSRF-Token": tok})
+        self.assertEqual(r.status_code, 400)
+
+
+if __name__ == "__main__":
+    unittest.main()