Add authentication backend based on YAML file

Author: RealOrangeOne

code_submitter/auth.py

@@ -1,10 +1,12 @@
-import base64
-import logging
-import binascii
-from typing import cast, List, Tuple, Optional, Sequence
 from typing_extensions import TypedDict
 
 import httpx
+import base64
+import logging
+import secrets
+import binascii
+from typing import cast, Dict, List, Tuple, Optional, Sequence
+from ruamel.yaml import YAML
 from starlette.requests import HTTPConnection
 from starlette.responses import Response
 from starlette.applications import Starlette
@@ -199,3 +201,45 @@ def __init__(self, data: List[NemesisUserInfo] = DEFAULT) -> None:
 
     async def load_user(self, username: str, password: str) -> NemesisUserInfo:
         return self.data[username]
+
+
+class FileBackend(BasicAuthBackend):
+    """
+    Authentication backend which stores credentials in a YAML file.
+
+    Credentials are stored in the format `TLA: password`.
+    """
+
+    UNKNOWN_USER_MESSAGE = "Username or password is incorrect"
+    BLUESHIRT_TEAM = "SRX"
+
+    def __init__(
+        self,
+        *,
+        path: str,
+    ) -> None:
+        with open(path) as f:
+            self.credentials = cast(Dict[str, str], YAML(typ="safe").load(f))
+
+    def get_scopes(self, username: str) -> List[str]:
+        scopes = ['authenticated']
+
+        if username == self.BLUESHIRT_TEAM:
+            scopes.append('blueshirt')
+
+        return scopes
+
+    async def validate(self, username: str, password: str) -> ValidationResult:
+        known_password = self.credentials.get(username)
+
+        if known_password is None:
+            raise AuthenticationError(self.UNKNOWN_USER_MESSAGE)
+
+        if not secrets.compare_digest(password.encode(), known_password.encode()):
+            raise AuthenticationError(self.UNKNOWN_USER_MESSAGE)
+
+        scopes = self.get_scopes(username)
+
+        if 'blueshirt' in scopes:
+            return scopes, User("SR", None)
+        return scopes, User(f"Team {username}", username)

requirements.in

@@ -5,3 +5,4 @@ databases[sqlite]
 sqlalchemy
 alembic
 httpx
+ruamel.yaml

requirements.txt

@@ -27,6 +27,8 @@ python-dateutil==2.8.1    # via alembic
 python-editor==1.0.4      # via alembic
 python-multipart==0.0.5   # via -r requirements.in
 rfc3986==1.4.0            # via httpx
+ruamel.yaml.clib==0.2.2   # via ruamel.yaml
+ruamel.yaml==0.16.12      # via -r requirements.in
 six==1.15.0               # via python-dateutil, python-multipart
 sniffio==1.1.0            # via httpcore, httpx
 sqlalchemy==1.3.18        # via -r requirements.in, alembic, databases

tests/fixtures/auth-file.yml

@@ -0,0 +1,2 @@
+ABC: password1
+SRX: bees

tests/tests_auth.py

@@ -1,10 +1,12 @@
+import os
 import test_utils
 from starlette.requests import Request
-from code_submitter.auth import NemesisBackend, NemesisUserInfo
 from starlette.responses import Response, JSONResponse
 from starlette.applications import Starlette
 from starlette.authentication import AuthenticationError
 
+from code_submitter.auth import FileBackend, NemesisBackend, NemesisUserInfo
+
 
 class NemesisAuthTests(test_utils.AsyncTestCase):
     def setUp(self) -> None:
@@ -92,3 +94,43 @@ async def endpoint(request: Request) -> Response:
             scopes,
             "Wrong scopes for user",
         )
+
+
+class FileAuthTests(test_utils.AsyncTestCase):
+    def setUp(self) -> None:
+        super().setUp()
+
+        my_dir = os.path.abspath(os.path.dirname(__file__))
+
+        self.backend = FileBackend(path=os.path.join(my_dir, "fixtures", "auth-file.yml"))
+
+    def test_ok(self) -> None:
+        scopes, user = self.await_(self.backend.validate('ABC', 'password1'))
+        self.assertEqual(['authenticated'], scopes, "Wrong scopes for user")
+
+        self.assertEqual('Team ABC', user.username, "Wrong username for user")
+        self.assertEqual('ABC', user.team, "Wrong team for user")
+
+    def test_unknown_user(self) -> None:
+        with self.assertRaises(AuthenticationError) as e:
+            self.await_(self.backend.validate('DEF', 'password1'))
+
+        self.assertEqual(e.exception.args[0], FileBackend.UNKNOWN_USER_MESSAGE)
+
+    def test_incorrect_password(self) -> None:
+        with self.assertRaises(AuthenticationError) as e:
+            self.await_(self.backend.validate('ABC', 'password2'))
+
+        self.assertEqual(e.exception.args[0], FileBackend.UNKNOWN_USER_MESSAGE)
+
+    def test_blueshirt(self) -> None:
+        scopes, user = self.await_(self.backend.validate('SRX', 'bees'))
+
+        self.assertIsNone(user.team, "Wrong team for user")
+        self.assertEqual('SR', user.username, "Wrong username for user")
+
+        self.assertEqual(
+            ['authenticated', 'blueshirt'],
+            scopes,
+            "Wrong scopes for user",
+        )