Reverting #2099

app/api_key/rest.py

@@ -121,7 +121,8 @@ def revoke_api_keys():
 
     # Step 1
     try:
-        api_key_token = api_key_data["token"]
+        # 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 = get_api_key_by_secret(api_key_token)
     except Exception:
         current_app.logger.error(

app/authentication/auth.py

@@ -152,12 +152,20 @@ 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)
-    except ValueError:
-        raise AuthError("Invalid token: Enter your full API key", 403)
     _auth_with_api_key(api_key, api_key.service)
 
 

app/dao/api_key_dao.py

@@ -83,30 +83,13 @@ 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, 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 ValueError()
-
-    # 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))
+def get_api_key_by_secret(secret):
+    signed_with_all_keys = signer_api_key.sign_with_all_keys(str(secret))
     for signed_secret in signed_with_all_keys:
         try:
-            api_key = db.on_reader().query(ApiKey).filter_by(_secret=signed_secret).options(joinedload("service")).one()
+            return db.on_reader().query(ApiKey).filter_by(_secret=signed_secret).options(joinedload("service")).one()
         except NoResultFound:
-            raise NoResultFound()
-
-    # Check the middle portion of the secret is the valid service id
-    if api_key and 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 ValueError()
-        else:
-            raise ValueError()
-    if api_key:
-        return api_key
+            pass
     raise NoResultFound()
 
 

tests/app/api_key/test_rest.py

@@ -81,9 +81,6 @@ 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"),
@@ -92,7 +89,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(unsigned_secret)
+        api_key_1 = get_api_key_by_secret(api_key_1.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"

tests/app/authentication/test_authentication.py

@@ -135,46 +135,41 @@ 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)
-    unsigned_secret = f"gcntfy-keyname-{sample_api_key.service_id}-{api_key_secret}"
-    response = client.get("/notifications", headers={"Authorization": f"{scheme} {unsigned_secret}"})
+    response = client.get("/notifications", headers={"Authorization": f"{scheme} {api_key_secret}"})
 
     assert response.status_code == 200
 
 
-def test_should_allow_auth_with_api_key_scheme_and_extra_spaces(client, sample_api_key):
-    api_key_secret = get_unsigned_secret(sample_api_key.id)
-    unsigned_secret = f"gcntfy-keyname-{sample_api_key.service_id}-{api_key_secret}"
-    response = client.get("/notifications", headers={"Authorization": f"ApiKey-v1    {unsigned_secret}"})
+# def test_should_allow_auth_with_api_key_scheme_and_extra_spaces(client, sample_api_key):
+#     api_key_secret = get_unsigned_secret(sample_api_key.id)
+#     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 == 200
+#     assert response.status_code == 200
 
 
-def test_should_NOT_allow_auth_with_api_key_scheme_with_incorrect_format(client, sample_api_key):
+def test_should_allow_auth_with_api_key_scheme_36_chars_or_longer(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 == 403
-    error_message = json.loads(response.get_data())
-    assert error_message["message"] == {"token": ["Invalid token: Enter your full API key"]}
+    assert response.status_code == 200
 
 
 def test_should_not_allow_invalid_api_key(client, sample_api_key):
     response = client.get("/notifications", headers={"Authorization": "ApiKey-v1 nope"})
 
     assert response.status_code == 403
     error_message = json.loads(response.get_data())
-    assert error_message["message"] == {"token": ["Invalid token: Enter your full API key"]}
+    assert error_message["message"] == {"token": ["Invalid token: API key not found"]}
 
 
 def test_should_not_allow_expired_api_key(client, sample_api_key):
     api_key_secret = get_unsigned_secret(sample_api_key.id)
 
     expire_api_key(service_id=sample_api_key.service_id, api_key_id=sample_api_key.id)
 
-    unsigned_secret = f"gcntfy-keyname-{sample_api_key.service_id}-{api_key_secret}"
-
-    response = client.get("/notifications", headers={"Authorization": f"ApiKey-v1 {unsigned_secret}"})
+    response = client.get("/notifications", headers={"Authorization": f"ApiKey-v1 {api_key_secret}"})
 
     assert response.status_code == 403
     error_message = json.loads(response.get_data())

tests/app/dao/test_api_key_dao.py

@@ -116,27 +116,12 @@ def test_get_unsigned_secret_returns_key(sample_api_key):
     assert unsigned_api_key == sample_api_key.secret
 
 
-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(ValueError):
-            get_api_key_by_secret("nope")
-
-        # Test getting secret without the keyname prefix
-        with pytest.raises(ValueError):
-            get_api_key_by_secret(str(sample_api_key.id))
-
-        # Test the service_name isnt part of the secret
-        with pytest.raises(ValueError):
-            get_api_key_by_secret(f"gcntfy-keyname-hello-{secret}")
-
-        # Test the secret is incorrect
-        with pytest.raises(NoResultFound):
-            get_api_key_by_secret(f"gcntfy-keyname-hello-{sample_api_key.service_id}-1234")
+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")
 
 
 def test_should_not_allow_duplicate_key_names_per_service(sample_api_key, fake_uuid):

tests/app/v2/notifications/test_post_notifications.py

@@ -1515,16 +1515,14 @@ 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 {unsigned_secret}"),
+                        ("Authorization", f"ApiKey-v1 {get_unsigned_secret(api_key.id)}"),
                     ],
                 )
                 return response
@@ -1565,16 +1563,14 @@ 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 {unsigned_secret}"),
+                        ("Authorization", f"ApiKey-v1 {get_unsigned_secret(api_key.id)}"),
                     ],
                 )
                 return response
@@ -1611,16 +1607,14 @@ 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 {unsigned_secret}"),
+                        ("Authorization", f"ApiKey-v1 {get_unsigned_secret(api_key.id)}"),
                     ],
                 )
                 return response