Enforce reasonable limits on API keys and Sessions per Principal

tiled/_tests/test_authentication.py

@@ -11,6 +11,7 @@
 from ..client import from_config
 from ..client.context import CannotRefreshAuthentication
 from ..client.utils import ClientError
+from ..server import authentication
 
 arr = ArrayAdapter.from_array(numpy.ones((5, 5)))
 
@@ -328,3 +329,35 @@ def test_api_keys(enter_password, config):
     time.sleep(2)
     with fail_with_status_code(401):
         from_config(config, api_key=user_key_info["secret"])
+
+
+def test_api_key_limit(enter_password, config):
+    # Decrease the limit so this test runs faster.
+    original_limit = authentication.API_KEY_LIMIT
+    authentication.API_KEY_LIMIT = 10
+    try:
+        with enter_password("secret2"):
+            user_client = from_config(config, username="bob", token_cache={})
+
+        for i in range(authentication.API_KEY_LIMIT):
+            user_client.context.create_api_key(note=f"key {i}")
+        # Hit API key limit.
+        with fail_with_status_code(400):
+            user_client.context.create_api_key(note="one key too many")
+    finally:
+        authentication.API_KEY_LIMIT = original_limit
+
+
+def test_session_limit(enter_password, config):
+    # Decrease the limit so this test runs faster.
+    original_limit = authentication.SESSION_LIMIT
+    authentication.SESSION_LIMIT = 10
+    try:
+        with enter_password("secret1"):
+            for _ in range(authentication.SESSION_LIMIT):
+                from_config(config, username="alice", token_cache={})
+            # Hit Session limit.
+            with fail_with_status_code(400):
+                from_config(config, username="alice", token_cache={})
+    finally:
+        authentication.SESSION_LIMIT = original_limit

tiled/server/authentication.py

@@ -47,6 +47,14 @@
 ALGORITHM = "HS256"
 UNIT_SECOND = timedelta(seconds=1)
 
+# Max API keys and Sessions allowed to Principal.
+# This is here for at least two reasons:
+# 1. Ensure that the routes which list API keys and sessions, which are
+#    not paginated, returns in a reasonable time.
+# 2. Avoid unintentional or intentional abuse.
+API_KEY_LIMIT = 100
+SESSION_LIMIT = 200
+
 
 def utcnow():
     "UTC now with second resolution"
@@ -336,6 +344,19 @@ def create_session(settings, identity_provider, id):
         else:
             identity.latest_login = now
             principal = identity.principal
+        session_count = (
+            db.query(orm.Session)
+            .join(orm.Principal)
+            .filter(orm.Principal.id == principal.id)
+            .count()
+        )
+        if session_count >= SESSION_LIMIT:
+            raise HTTPException(
+                400,
+                f"This Principal already has {session_count} sessions which is greater "
+                f"than or equal to the maximum number allowed, {SESSION_LIMIT}. "
+                "Some Sessions must be closed before creating new ones.",
+            )
         session = orm.Session(
             principal_id=principal.id,
             expiration_time=utcnow() + settings.session_max_age,
@@ -444,6 +465,19 @@ def generate_apikey(db, principal, apikey_params, request):
     # plus 4 more for extra safety since we store the first eight HEX chars.
     secret = secrets.token_bytes(4 + 32)
     hashed_secret = hashlib.sha256(secret).digest()
+    keys_count = (
+        db.query(orm.APIKey)
+        .join(orm.Principal)
+        .filter(orm.Principal.id == principal.id)
+        .count()
+    )
+    if keys_count >= API_KEY_LIMIT:
+        raise HTTPException(
+            400,
+            f"This Principal already has {keys_count} API keys which is greater "
+            f"than or equal to the maximum number allowed, {API_KEY_LIMIT}. "
+            "Some API keys must be deleted before creating new ones.",
+        )
     new_key = orm.APIKey(
         principal_id=principal.id,
         expiration_time=expiration_time,