Add API Validation (#2081)

* validate apli keys * Add api key validation - We check the API Prefix - We check that the service_id is in the api key - Else return an error * Test changes * fix errors * fix errors * Update tests/app/authentication/test_authentication.py Co-authored-by: Andrew <[email protected]> --------- Co-authored-by: Andrew <[email protected]>
cds-snc · Jan 18, 2024 · 0d25f17 · 0d25f17
1 parent 92febf2
commit 0d25f17
Show file tree

Hide file tree

Showing 7 changed files with 61 additions and 31 deletions.
diff --git a/app/api_key/rest.py b/app/api_key/rest.py
@@ -121,8 +121,7 @@ def revoke_api_keys():
 
     # Step 1
     try:
-        # take last 36 chars of string so that it works even if the full key is provided.
-        api_key_token = api_key_data["token"][-36:]
+        api_key_token = api_key_data["token"]
         api_key = get_api_key_by_secret(api_key_token)
     except Exception:
         current_app.logger.error(

diff --git a/app/authentication/auth.py b/app/authentication/auth.py
@@ -152,19 +152,8 @@ def requires_auth():
 
 
 def _auth_by_api_key(auth_token):
-    # TODO: uncomment this when the grace period for the token prefix is over
-    # orig_token = auth_token
-
     try:
-        # take last 36 chars of string so that it works even if the full key is provided.
-        auth_token = auth_token[-36:]
         api_key = get_api_key_by_secret(auth_token)
-
-        # TODO: uncomment this when the grace period for the token prefix is over
-        # check for token prefix
-        # if current_app.config["API_KEY_PREFIX"] not in orig_token:
-        #     raise AuthError("Invalid token: you must re-generate your API key to continue using GC Notify", 403, service_id=api_key.service.id, api_key_id=api_key.id)
-
     except NoResultFound:
         raise AuthError("Invalid token: API key not found", 403)
     _auth_with_api_key(api_key, api_key.service)

diff --git a/app/dao/api_key_dao.py b/app/dao/api_key_dao.py
@@ -76,13 +76,30 @@ def update_compromised_api_key_info(service_id, api_key_id, compromised_info):
     db.session.add(api_key)
 
 
-def get_api_key_by_secret(secret):
-    signed_with_all_keys = signer_api_key.sign_with_all_keys(str(secret))
+def get_api_key_by_secret(secret, service_id=None):
+    # Check the first part of the secret is the gc prefix
+    if current_app.config["API_KEY_PREFIX"] != secret[: len(current_app.config["API_KEY_PREFIX"])]:
+        raise NoResultFound()
+
+    # Check if the remaining part of the secret is a the valid api key
+    token = secret[-36:]
+    signed_with_all_keys = signer_api_key.sign_with_all_keys(str(token))
     for signed_secret in signed_with_all_keys:
         try:
-            return db.on_reader().query(ApiKey).filter_by(_secret=signed_secret).options(joinedload("service")).one()
+            api_key = db.on_reader().query(ApiKey).filter_by(_secret=signed_secret).options(joinedload("service")).one()
         except NoResultFound:
             pass
+
+    # Check the middle portion of the secret is the valid service id
+    if api_key.service_id:
+        if len(secret) >= 79:
+            service_id_from_token = str(secret[-73:-37])
+            if str(api_key.service_id) != service_id_from_token:
+                raise NoResultFound()
+        else:
+            raise NoResultFound()
+    if api_key:
+        return api_key
     raise NoResultFound()
 
 

diff --git a/tests/app/api_key/test_rest.py b/tests/app/api_key/test_rest.py
@@ -90,6 +90,9 @@ def test_revoke_api_keys_with_valid_auth_revokes_and_notifies_user(self, client,
         api_key_1 = create_api_key(service, key_type=KEY_TYPE_NORMAL, key_name="Key 1")
         unsigned_secret = get_unsigned_secret(api_key_1.id)
 
+        # Create token expected from the frontend
+        unsigned_secret = f"gcntfy-keyname-{service.id}-{unsigned_secret}"
+
         sre_auth_header = create_sre_authorization_header()
         response = client.post(
             url_for("sre_tools.revoke_api_keys"),
@@ -98,7 +101,7 @@ def test_revoke_api_keys_with_valid_auth_revokes_and_notifies_user(self, client,
         )
 
         # Get api key from DB
-        api_key_1 = get_api_key_by_secret(api_key_1.secret)
+        api_key_1 = get_api_key_by_secret(unsigned_secret)
         assert response.status_code == 201
         assert api_key_1.expiry_date is not None
         assert api_key_1.compromised_key_info["type"] == "cds-tester"

diff --git a/tests/app/authentication/test_authentication.py b/tests/app/authentication/test_authentication.py
@@ -135,18 +135,20 @@ def test_admin_auth_should_not_allow_api_key_scheme(client, sample_api_key):
 @pytest.mark.parametrize("scheme", ["ApiKey-v1", "apikey-v1", "APIKEY-V1"])
 def test_should_allow_auth_with_api_key_scheme(client, sample_api_key, scheme):
     api_key_secret = get_unsigned_secret(sample_api_key.id)
-
-    response = client.get("/notifications", headers={"Authorization": f"{scheme} {api_key_secret}"})
+    unsigned_secret = f"gcntfy-keyname-{sample_api_key.service_id}-{api_key_secret}"
+    response = client.get("/notifications", headers={"Authorization": f"{scheme} {unsigned_secret}"})
 
     assert response.status_code == 200
 
 
-def test_should_allow_auth_with_api_key_scheme_36_chars_or_longer(client, sample_api_key):
+def test_should_NOT_allow_auth_with_api_key_scheme_with_incorrect_format(client, sample_api_key):
     api_key_secret = "fhsdkjhfdsfhsd" + get_unsigned_secret(sample_api_key.id)
 
     response = client.get("/notifications", headers={"Authorization": f"ApiKey-v1 {api_key_secret}"})
 
-    assert response.status_code == 200
+    assert response.status_code == 403
+    error_message = json.loads(response.get_data())
+    assert error_message["message"] == {"token": ["Invalid token: API key not found"]}
 
 
 def test_should_not_allow_invalid_api_key(client, sample_api_key):
@@ -162,7 +164,9 @@ def test_should_not_allow_expired_api_key(client, sample_api_key):
 
     expire_api_key(service_id=sample_api_key.service_id, api_key_id=sample_api_key.id)
 
-    response = client.get("/notifications", headers={"Authorization": f"ApiKey-v1 {api_key_secret}"})
+    unsigned_secret = f"gcntfy-keyname-{sample_api_key.service_id}-{api_key_secret}"
+
+    response = client.get("/notifications", headers={"Authorization": f"ApiKey-v1 {unsigned_secret}"})
 
     assert response.status_code == 403
     error_message = json.loads(response.get_data())

diff --git a/tests/app/dao/test_api_key_dao.py b/tests/app/dao/test_api_key_dao.py
@@ -103,12 +103,24 @@ def test_get_unsigned_secret_returns_key(sample_api_key):
     assert unsigned_api_key == sample_api_key.secret
 
 
-def test_get_api_key_by_secret(sample_api_key):
-    unsigned_secret = get_unsigned_secret(sample_api_key.id)
-    assert get_api_key_by_secret(unsigned_secret).id == sample_api_key.id
-
-    with pytest.raises(NoResultFound):
-        get_api_key_by_secret("nope")
+class TestGetAPIKeyBySecret:
+    def test_get_api_key_by_secret(self, sample_api_key):
+        secret = get_unsigned_secret(sample_api_key.id)
+        # Create token expected from the frontend
+        unsigned_secret = f"gcntfy-keyname-{sample_api_key.service_id}-{secret}"
+        assert get_api_key_by_secret(unsigned_secret).id == sample_api_key.id
+
+        with pytest.raises(NoResultFound):
+            get_api_key_by_secret("nope")
+
+        # Test getting secret without the keyname prefix
+        with pytest.raises(NoResultFound):
+            get_api_key_by_secret(str(sample_api_key.id))
+
+        # Test the service_name isnt part of the secret
+        with pytest.raises(NoResultFound):
+            # import pdb; pdb.set_trace()
+            get_api_key_by_secret(f"gcntfy-keyname-hello-{secret}")
 
 
 def test_should_not_allow_duplicate_key_names_per_service(sample_api_key, fake_uuid):

diff --git a/tests/app/v2/notifications/test_post_notifications.py b/tests/app/v2/notifications/test_post_notifications.py
@@ -1515,14 +1515,16 @@ def __send_sms():
                 key_type=key_type,
             )
             save_model_api_key(api_key)
+            api_key_secret = get_unsigned_secret(api_key.id)
+            unsigned_secret = f"gcntfy-keyname-{api_key.service_id}-{api_key_secret}"
 
             with set_config_values(notify_api, {"REDIS_ENABLED": True}):
                 response = client.post(
                     path="/v2/notifications/sms",
                     data=json.dumps(data),
                     headers=[
                         ("Content-Type", "application/json"),
-                        ("Authorization", f"ApiKey-v1 {get_unsigned_secret(api_key.id)}"),
+                        ("Authorization", f"ApiKey-v1 {unsigned_secret}"),
                     ],
                 )
                 return response
@@ -1563,14 +1565,16 @@ def __send_sms():
                 key_type=key_type,
             )
             save_model_api_key(api_key)
+            api_key_secret = get_unsigned_secret(api_key.id)
+            unsigned_secret = f"gcntfy-keyname-{api_key.service_id}-{api_key_secret}"
 
             with set_config_values(notify_api, {"REDIS_ENABLED": True}):
                 response = client.post(
                     path="/v2/notifications/bulk",
                     data=json.dumps(data),
                     headers=[
                         ("Content-Type", "application/json"),
-                        ("Authorization", f"ApiKey-v1 {get_unsigned_secret(api_key.id)}"),
+                        ("Authorization", f"ApiKey-v1 {unsigned_secret}"),
                     ],
                 )
                 return response
@@ -1607,14 +1611,16 @@ def __send_sms():
                 key_type=key_type,
             )
             save_model_api_key(api_key)
+            api_key_secret = get_unsigned_secret(api_key.id)
+            unsigned_secret = f"gcntfy-keyname-{api_key.service_id}-{api_key_secret}"
 
             with set_config_values(notify_api, {"REDIS_ENABLED": True}):
                 response = client.post(
                     path="/v2/notifications/bulk",
                     data=json.dumps(data),
                     headers=[
                         ("Content-Type", "application/json"),
-                        ("Authorization", f"ApiKey-v1 {get_unsigned_secret(api_key.id)}"),
+                        ("Authorization", f"ApiKey-v1 {unsigned_secret}"),
                     ],
                 )
                 return response