increase test coverage for vault and login api

.gitignore

@@ -5,3 +5,5 @@ package-lock.json
 # mkcert certificates
 *.pem
 venv/
+
+bootstrap-alpha.yml

accounts/accounts/api.py

@@ -1,6 +1,5 @@
 from datetime import datetime, timedelta
 from os import environ
-import base64
 from functools import wraps
 
 from flask import jsonify, make_response, request
@@ -11,6 +10,8 @@ import jwt
 from accounts import app
 from accounts.models import User
 
+COOKIE_KEY = "auth"
+
 
 def json_error(handler):
     @wraps(handler)
@@ -48,7 +49,7 @@ def post_login():
             raise UnauthorizedError("bad password")
     except UnauthorizedError as unauthorized_error:
         resp = make_response(jsonify({"error": str(unauthorized_error), "status": 401}))
-        resp.set_cookie("auth", "", expires=0)
+        resp.set_cookie(COOKIE_KEY, "", expires=0)
         resp.status_code = 401
         return resp
 
@@ -56,7 +57,6 @@ def post_login():
     expiry = datetime.utcnow() + timedelta(hours=24)
     encoded = jwt.encode(
         {
-            "ok": True,
             "exp": expiry,
             "priv": {
                 "userId": match.user_id,
@@ -69,7 +69,7 @@ def post_login():
 
     resp = make_response(jsonify({"user": match.serialize()}))
     resp.set_cookie(
-        "auth",
+        COOKIE_KEY,
         encoded,
         httponly=True,
         expires=expiry,
@@ -84,7 +84,7 @@ def post_login():
 @cross_origin(origins=[environ.get("CORS_ORIGIN", "*")], supports_credentials=True)
 @json_error
 def get_login():
-    auth_cookie = request.cookies.get("auth")
+    auth_cookie = request.cookies.get(COOKIE_KEY)
     public_key = environ.get("JWT_PUBLIC_KEY", "")
     try:
         token = jwt.decode(auth_cookie, public_key)
@@ -103,10 +103,22 @@ def get_login():
     return jsonify({"user": match.serialize()})
 
 
-# This route is not supposed to be called by client-side applications, so
-# no CORS configuration is added
+@app.route("/api/logout", methods=["POST"])
+@cross_origin(origins=[environ.get("CORS_ORIGIN", "*")], supports_credentials=True)
+@json_error
+def post_logout():
+    resp = make_response("")
+    resp.set_cookie(COOKIE_KEY, "", expires=0)
+    resp.status_code = 204
+    return resp
+
+
 @app.route("/api/key", methods=["GET"])
 @json_error
 def key():
+    """
+    This route is not supposed to be called by client-side applications, so
+    no CORS configuration is added
+    """
     public_key = environ.get("JWT_PUBLIC_KEY", "").strip()
     return jsonify({"key": public_key})

accounts/accounts/views.py

@@ -67,13 +67,15 @@ class UserView(ModelView):
     column_display_all_relations = True
     column_list = ("email", "user_id")
     form_columns = ("email", "accounts")
+    form_create_rules = ("email", "password", "confirm", "accounts")
+    form_edit_rules = ("email", "password", "confirm", "accounts")
 
     def on_model_change(self, form, model, is_created):
         if form.password.data:
             model.hashed_password = bcrypt.hash(form.password.data)
 
     def get_create_form(self):
-        form = super().get_create_form()
+        form = super(UserView, self).get_create_form()
         form.password = PasswordField(
             "Password",
             validators=[
@@ -85,13 +87,11 @@ class UserView(ModelView):
         return form
 
     def get_edit_form(self):
-        form = super().get_edit_form()
+        form = super(UserView, self).get_edit_form()
         form.password = PasswordField(
             "Password",
             description="When left blank, the password will remain unchanged on update",
-            validators=[
-                EqualTo("confirm", message="Passwords must match"),
-            ],
+            validators=[EqualTo("confirm", message="Passwords must match")],
         )
         form.confirm = PasswordField("Repeat Password", validators=[])
         return form

accounts/authorizer/__init__.py

@@ -17,6 +17,7 @@ def build_api_arn(method_arn):
         aws_region, aws_account_id, rest_api_id, stage
     )
 
+
 def build_response(api_arn, allow):
     effect = "Deny"
     if allow:
@@ -30,17 +31,18 @@ def build_response(api_arn, allow):
                 {
                     "Action": ["execute-api:Invoke"],
                     "Effect": effect,
-                    "Resource": [api_arn]
+                    "Resource": [api_arn],
                 }
-            ]
-        }
+            ],
+        },
     }
 
 
 def handler(event, context):
     api_arn = build_api_arn(event["methodArn"])
 
-    auth_string = base64.standard_b64decode(event["authorizationToken"].lstrip("Basic ")).decode()
+    encoded_auth = event["authorizationToken"].lstrip("Basic ")
+    auth_string = base64.standard_b64decode(encoded_auth).decode()
     if not auth_string:
         return build_response(api_arn, False)
 
@@ -51,7 +53,8 @@ def handler(event, context):
     if user != environ.get("BASIC_AUTH_USER"):
         return build_response(api_arn, False)
 
-    hashed_password = base64.standard_b64decode(environ.get("HASHED_BASIC_AUTH_PASSWORD")).decode()
+    encoded_password = environ.get("HASHED_BASIC_AUTH_PASSWORD")
+    hashed_password = base64.standard_b64decode(encoded_password).decode()
     if not bcrypt.verify(password, hashed_password):
         return build_response(api_arn, False)
 

accounts/serverless.yml

@@ -85,13 +85,13 @@ functions:
     environment:
       CORS_ORIGIN: https://${self:custom.origin.${self:custom.stage}}
       COOKIE_DOMAIN: ${self:custom.cookieDomain.${self:custom.stage}}
+      SERVER_URL: ${self:custom.serverHost.${self:custom.stage}}
       JWT_PRIVATE_KEY: '${ssm:/aws/reference/secretsmanager/${self:custom.stage}/accounts/jwtPrivateKey~true}'
       JWT_PUBLIC_KEY: '${ssm:/aws/reference/secretsmanager/${self:custom.stage}/accounts/jwtPublicKey~true}'
-      HASHED_BASIC_AUTH_PASSWORD: ${ssm:/aws/reference/secretsmanager/${self:custom.stage}/accounts/hashedBasicAuthPassword~true}
       BASIC_AUTH_USER: ${ssm:/aws/reference/secretsmanager/${self:custom.stage}/accounts/basicAuthUser~true}
+      HASHED_BASIC_AUTH_PASSWORD: ${ssm:/aws/reference/secretsmanager/${self:custom.stage}/accounts/hashedBasicAuthPassword~true}
       SESSION_SECRET: '${ssm:/aws/reference/secretsmanager/${self:custom.stage}/accounts/sessionSecret~true}'
-      SERVER_URL: ${self:custom.serverHost.${self:custom.stage}}
-      POSTGRES_CONNECTION_STRING: '${ssm:/aws/reference/secretsmanager/${self:custom.stage}/accounts/postgresConnectionString~true}'
+      MYSQL_CONNECTION_STRING: '${ssm:/aws/reference/secretsmanager/${self:custom.stage}/accounts/mysqlConnectionString~true}'
 
 resources:
   Resources:

accounts/tests/test_api.py

@@ -0,0 +1,148 @@
+import unittest
+import json
+import base64
+from json import loads
+from time import time
+from os import environ
+
+from accounts import app
+
+
+def _pad_b64_string(s):
+    while len(s) % 4 is not 0:
+        s = s + "="
+    return s
+
+
+class TestKey(unittest.TestCase):
+    def setUp(self):
+        self.app = app.test_client()
+
+    def test_get_key(self):
+        rv = self.app.get("/api/key")
+        assert rv.status.startswith("200")
+        data = loads(rv.data)
+        assert data["key"] == environ.get("JWT_PUBLIC_KEY")
+
+
+class TestJWT(unittest.TestCase):
+    def setUp(self):
+        self.app = app.test_client()
+
+    def _assert_cookie_present(self, name):
+        for cookie in self.app.cookie_jar:
+            if cookie.name == name:
+                return cookie.value
+        raise AssertionError("Cookie named {} not found".format(name))
+
+    def _assert_cookie_not_present(self, name):
+        for cookie in self.app.cookie_jar:
+            assert cookie.name != name
+
+    def test_jwt_flow(self):
+        """
+        First, try login attempts that are supposed to fail:
+        1. checking login status without any prior interaction
+        2. try logging in with an unknown user
+        3. try logging in with a known user and bad password
+        """
+        rv = self.app.get("/api/login")
+        assert rv.status.startswith("401")
+        self._assert_cookie_not_present("auth")
+
+        rv = self.app.post(
+            "/api/login",
+            data=json.dumps(
+                {"username": "does@not.exist", "password": "somethingsomething"}
+            ),
+        )
+        assert rv.status.startswith("401")
+        self._assert_cookie_not_present("auth")
+
+        rv = self.app.post(
+            "/api/login",
+            data=json.dumps({"username": "develop@offen.dev", "password": "developp"}),
+        )
+        assert rv.status.startswith("401")
+        self._assert_cookie_not_present("auth")
+
+        """
+        Next, perform a successful login
+        """
+        rv = self.app.post(
+            "/api/login",
+            data=json.dumps({"username": "develop@offen.dev", "password": "develop"}),
+        )
+        assert rv.status.startswith("200")
+
+        """
+        The response should contain information about the
+        user and full information (i.e. a name) about the associated accounts
+        """
+        data = json.loads(rv.data)
+        assert data["user"]["userId"] is not None
+        self.assertListEqual(
+            data["user"]["accounts"],
+            [
+                {"name": "One", "accountId": "9b63c4d8-65c0-438c-9d30-cc4b01173393"},
+                {"name": "Two", "accountId": "78403940-ae4f-4aff-a395-1e90f145cf62"},
+            ],
+        )
+
+        """
+        The claims part of the JWT is expected to contain a valid expiry,
+        information about the user and the associated account ids.
+        """
+        jwt = self._assert_cookie_present("auth")
+        # PyJWT strips the padding from the base64 encoded parts which Python
+        # cannot decode properly, so we need to add the padding ourselves
+        claims_part = _pad_b64_string(jwt.split(".")[1])
+        claims = loads(base64.b64decode(claims_part))
+        assert claims.get("exp") > time()
+
+        priv = claims.get("priv")
+        assert priv is not None
+
+        assert priv.get("userId") is not None
+        self.assertListEqual(
+            priv["accounts"],
+            [
+                "9b63c4d8-65c0-438c-9d30-cc4b01173393",
+                "78403940-ae4f-4aff-a395-1e90f145cf62",
+            ],
+        )
+
+        """
+        Checking the login status when re-using the cookie should yield
+        a successful response
+        """
+        rv = self.app.get("/api/login")
+        assert rv.status.startswith("200")
+        jwt2 = self._assert_cookie_present("auth")
+        assert jwt2 == jwt
+
+        """
+        Performing a bad login attempt when sending a valid auth cookie
+        is expected to destroy the cookie and leave the user logged out again
+        """
+        rv = self.app.post(
+            "/api/login",
+            data=json.dumps(
+                {"username": "evil@session.takeover", "password": "develop"}
+            ),
+        )
+        assert rv.status.startswith("401")
+        self._assert_cookie_not_present("auth")
+
+        """
+        Explicitly logging out leaves the user without cookies
+        """
+        rv = self.app.post(
+            "/api/login",
+            data=json.dumps({"username": "develop@offen.dev", "password": "develop"}),
+        )
+        assert rv.status.startswith("200")
+
+        rv = self.app.post("/api/logout")
+        assert rv.status.startswith("204")
+        self._assert_cookie_not_present("auth")

accounts/tests/test_jwt.py

@@ -1,21 +0,0 @@
-import unittest
-import json
-
-from accounts import app
-
-
-class TestJWT(unittest.TestCase):
-    def setUp(self):
-        self.app = app.test_client()
-
-    def test_jwt_flow(self):
-        rv = self.app.get("/api/login")
-        assert rv.status.startswith("401")
-
-        rv = self.app.post(
-            "/api/login", data=json.dumps({"username": "develop@offen.dev", "password": "develop"})
-        )
-        assert rv.status.startswith("200")
-
-        rv = self.app.get("/api/login")
-        assert rv.status.startswith("200")