Implement a nemesis auth backend

Fixes #1.

Author: PeterJCLaw

code_submitter/auth.py

@@ -1,16 +1,22 @@
 import base64
+import logging
 import binascii
-from typing import Tuple, Optional, Sequence
+from typing import cast, List, Tuple, Optional, Sequence
+from typing_extensions import TypedDict
 
+import httpx
 from starlette.requests import HTTPConnection
 from starlette.responses import Response
+from starlette.applications import Starlette
 from starlette.authentication import (
     SimpleUser,
     AuthCredentials,
     AuthenticationError,
     AuthenticationBackend,
 )
 
+logger = logging.getLogger(__name__)
+
 
 class User(SimpleUser):
     def __init__(self, username: str, team: Optional[str]) -> None:
@@ -70,3 +76,81 @@ async def validate(self, username: str, password: str) -> ValidationResult:
         if not password:
             raise AuthenticationError("Must provide a password")
         return ['authenticated'], User(username, self.team)
+
+
+NemesisUserInfo = TypedDict('NemesisUserInfo', {
+    'username': str,
+    'first_name': str,
+    'last_name': str,
+    'teams': List[str],
+    'is_blueshirt': bool,
+    'is_student': bool,
+    'is_team_leader': bool,
+})
+
+
+class NemesisBackend(BasicAuthBackend):
+    def __init__(self, _target: Optional[Starlette] = None, *, url: str) -> None:
+        # Munge types to cope with httpx not supporting strict_optional but
+        # actually being fine with given `None`. Note we expect only to pass
+        # this value in tests, so need to cope with it being `None` most of the
+        # time anyway.
+        app = cast(Starlette, _target)
+        self.client = httpx.AsyncClient(base_url=url, app=app)
+
+    async def load_user(self, username: str, password: str) -> NemesisUserInfo:
+        async with self.client as client:
+            respone = await client.get(
+                'user/{}'.format(username),
+                auth=(username, password),
+            )
+
+            try:
+                respone.raise_for_status()
+            except httpx.HTTPError as e:
+                if e.response.status_code != 403:
+                    logger.exception(
+                        "Failed to contact nemesis while trying to authenticate %r",
+                        username,
+                    )
+                raise AuthenticationError(e) from e
+
+            return cast(NemesisUserInfo, respone.json())
+
+    def strip_team(self, team: str) -> str:
+        # All teams from nemesis *should* start with this prefix...
+        if team.startswith('team-'):
+            return team[len('team-'):]
+        return team
+
+    def get_team(self, info: NemesisUserInfo) -> Optional[str]:
+        teams = [self.strip_team(x) for x in info['teams']]
+
+        if not teams:
+            if info['is_student']:
+                logger.warning("Competitor %r has no teams!", info['username'])
+            return None
+
+        team = teams[0]
+
+        if len(teams) > 1:
+            logger.warning(
+                "User %r is in more than one team (%r), using %r",
+                info['username'],
+                teams,
+                team,
+            )
+
+        return team
+
+    async def validate(self, username: str, password: str) -> ValidationResult:
+        if not username:
+            raise AuthenticationError("Must provide a username")
+        if not password:
+            raise AuthenticationError("Must provide a password")
+
+        info = await self.load_user(username, password)
+
+        team = self.get_team(info)
+
+        return ['authenticated'], User(username, team)

setup.py

@@ -43,5 +43,6 @@
         'databases[sqlite]',
         'sqlalchemy',
         'alembic',
+        'httpx',
     ],
 )

tests/tests_auth.py

@@ -0,0 +1,76 @@
+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
+
+
+class NemesisAuthTests(test_utils.AsyncTestCase):
+    def setUp(self) -> None:
+        super().setUp()
+
+        self.info = NemesisUserInfo({
+            'username': 'user',
+            'first_name': 'Dave',
+            'last_name': 'McDave',
+            'teams': ['team-ABC'],
+            'is_blueshirt': False,
+            'is_student': True,
+            'is_team_leader': False,
+        })
+        self.fake_nemesis = Starlette()
+        self.backend = NemesisBackend(self.fake_nemesis, url='http://nowhere/')
+
+    def test_not_authenticated(self) -> None:
+        @self.fake_nemesis.route('/user/user')
+        async def endpoint(request: Request) -> Response:
+            return JSONResponse(
+                {'authentication_errors': [
+                    'NO_USERNAME',
+                    'NO_PASSWORD',
+                    'WRONG_PASSWORD',
+                ]},
+                status_code=403,
+            )
+
+        with self.assertRaises(AuthenticationError):
+            self.await_(self.backend.validate('user', 'pass'))
+
+    def test_ok(self) -> None:
+        @self.fake_nemesis.route('/user/user')
+        async def endpoint(request: Request) -> Response:
+            return JSONResponse(self.info)
+
+        scopes, user = self.await_(self.backend.validate('user', 'pass'))
+
+        self.assertEqual(['authenticated'], scopes, "Wrong scopes for user")
+
+        self.assertEqual('user', user.username, "Wrong username for user")
+        self.assertEqual('ABC', user.team, "Wrong team for user")
+
+    def test_no_team(self) -> None:
+        @self.fake_nemesis.route('/user/user')
+        async def endpoint(request: Request) -> Response:
+            self.info['teams'] = []
+            return JSONResponse(self.info)
+
+        scopes, user = self.await_(self.backend.validate('user', 'pass'))
+
+        self.assertIsNone(user.team, "Wrong team for user")
+        self.assertEqual('user', user.username, "Wrong username for user")
+
+        self.assertEqual(['authenticated'], scopes, "Wrong scopes for user")
+
+    def test_multiple_teams(self) -> None:
+        @self.fake_nemesis.route('/user/user')
+        async def endpoint(request: Request) -> Response:
+            self.info['teams'] = ['team-DEF', 'team-ABC']
+            return JSONResponse(self.info)
+
+        scopes, user = self.await_(self.backend.validate('user', 'pass'))
+
+        self.assertEqual('DEF', user.team, "Wrong team for user")
+        self.assertEqual('user', user.username, "Wrong username for user")
+
+        self.assertEqual(['authenticated'], scopes, "Wrong scopes for user")