Merge branch 'file-auth-backend'

Author: PeterJCLaw

code_submitter/auth.py

@@ -1,9 +1,12 @@
 import base64
 import logging
+import secrets
 import binascii
-from typing import cast, List, Tuple, Optional, Sequence
+from typing import cast, Dict, List, Tuple, Union, Optional, Sequence
+from pathlib import Path
 from typing_extensions import TypedDict
 
+import yaml
 import httpx
 from starlette.requests import HTTPConnection
 from starlette.responses import Response
@@ -17,6 +20,8 @@
 
 logger = logging.getLogger(__name__)
 
+BLUESHIRT_SCOPE = 'blueshirt'
+
 
 class User(SimpleUser):
     def __init__(self, username: str, team: Optional[str]) -> None:
@@ -154,7 +159,7 @@ def get_scopes(self, info: NemesisUserInfo) -> List[str]:
         scopes = ['authenticated']
 
         if info['is_blueshirt']:
-            scopes.append('blueshirt')
+            scopes.append(BLUESHIRT_SCOPE)
 
         return scopes
 
@@ -199,3 +204,49 @@ 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`.
+
+    Note: Passwords are stored in plaintext.
+    This is acceptable for some use-cases, for example where the
+    data being uploaded is not sensitive or because someone having
+    access to the credentials file almost certainly implies they have
+    access to the upload storage anyway. However this may not be
+    the case for all use-cases and you should evaluate the risks
+    yourself before using this backend. You have been warned!
+    """
+
+    UNKNOWN_USER_MESSAGE = "Username or password is incorrect"
+    BLUESHIRT_TEAM = "SRX"
+
+    def __init__(self, *, path: Union[str, Path]) -> None:
+        with open(path) as f:
+            self.credentials = cast(Dict[str, str], yaml.safe_load(f))
+
+    def get_scopes(self, username: str) -> List[str]:
+        scopes = ['authenticated']
+
+        if username == self.BLUESHIRT_TEAM:
+            scopes.append(BLUESHIRT_SCOPE)
+
+        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_SCOPE in scopes:
+            return scopes, User("SR", None)
+        return scopes, User(f"Team {username}", username)

code_submitter/server.py

@@ -15,7 +15,7 @@
 from starlette.middleware.authentication import AuthenticationMiddleware
 
 from . import auth, utils, config
-from .auth import User
+from .auth import User, BLUESHIRT_SCOPE
 from .tables import Archive, ChoiceHistory
 
 database = databases.Database(config.DATABASE_URL, force_rollback=config.TESTING)
@@ -53,6 +53,7 @@ async def homepage(request: Request) -> Response:
         'request': request,
         'chosen': chosen,
         'uploads': uploads,
+        'BLUESHIRT_SCOPE': BLUESHIRT_SCOPE,
     })
 
 
@@ -121,7 +122,7 @@ async def upload(request: Request) -> Response:
     )
 
 
-@requires(['authenticated', 'blueshirt'])
+@requires(['authenticated', BLUESHIRT_SCOPE])
 async def download_submissions(request: Request) -> Response:
     buffer = io.BytesIO()
     with zipfile.ZipFile(buffer, mode='w') as zf:

requirements.in

@@ -5,3 +5,4 @@ databases[sqlite]
 sqlalchemy
 alembic
 httpx
+pyyaml

requirements.txt

@@ -26,6 +26,7 @@ markupsafe==1.1.1         # via jinja2, mako
 python-dateutil==2.8.1    # via alembic
 python-editor==1.0.4      # via alembic
 python-multipart==0.0.5   # via -r requirements.in
+pyyaml==5.3.1             # via -r requirements.in
 rfc3986==1.4.0            # via httpx
 six==1.15.0               # via python-dateutil, python-multipart
 sniffio==1.1.0            # via httpcore, httpx

templates/index.html

@@ -31,7 +31,7 @@
   <body>
     <div class="container">
       <h1>Virtual Competition Code Submission</h1>
-      {% if 'blueshirt' in request.auth.scopes %}
+      {% if BLUESHIRT_SCOPE in request.auth.scopes %}
       <div class="row">
         <div class="col-sm-6">
           <a href="{{ url_for('download_submissions') }}">

tests/fixtures/auth-file.yml

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

tests/tests_auth.py

@@ -1,10 +1,17 @@
+from pathlib import Path
+
 import test_utils
 from starlette.requests import Request
 from starlette.responses import Response, JSONResponse
 from starlette.applications import Starlette
 from starlette.authentication import AuthenticationError
 
-from code_submitter.auth import NemesisBackend, NemesisUserInfo
+from code_submitter.auth import (
+    FileBackend,
+    NemesisBackend,
+    BLUESHIRT_SCOPE,
+    NemesisUserInfo,
+)
 
 
 class NemesisAuthTests(test_utils.AsyncTestCase):
@@ -89,7 +96,47 @@ async def endpoint(request: Request) -> Response:
         self.assertEqual('user', user.username, "Wrong username for user")
 
         self.assertEqual(
-            ['authenticated', 'blueshirt'],
+            ['authenticated', BLUESHIRT_SCOPE],
+            scopes,
+            "Wrong scopes for user",
+        )
+
+
+class FileAuthTests(test_utils.AsyncTestCase):
+    def setUp(self) -> None:
+        super().setUp()
+
+        self.backend = FileBackend(
+            path=Path(__file__).parent / '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_SCOPE],
             scopes,
             "Wrong scopes for user",
         )