fix: truncate long User-Agent strings in user subscription updates

ImMohammad20000 · ImMohammad20000 · commit 08999a5a42ea · 2026-01-05T16:44:27.000+03:30
diff --git a/app/db/crud/user.py b/app/db/crud/user.py
@@ -32,6 +32,9 @@
 from .group import get_groups_by_ids
 
 
+_USER_AGENT_MAX_LEN = UserSubscriptionUpdate.__table__.columns.user_agent.type.length or 512
+
+
 async def load_user_attrs(user: User):
     await user.awaitable_attrs.admin
     await user.awaitable_attrs.next_plan
@@ -845,7 +848,9 @@ async def user_sub_update(db: AsyncSession, user_id: User, user_agent: str) -> U
         user_agent (str): The user agent string.
 
     """
-    agent = UserSubscriptionUpdate(user_id=user_id, user_agent=user_agent)
+    # Clamp to column length; some clients send very long strings (e.g. encoded configs) as User-Agent.
+    sanitized_user_agent = (user_agent or "")[:_USER_AGENT_MAX_LEN]
+    agent = UserSubscriptionUpdate(user_id=user_id, user_agent=sanitized_user_agent)
     db.add(agent)
     await db.commit()
 
diff --git a/tests/api/test_user.py b/tests/api/test_user.py
@@ -176,6 +176,29 @@ def test_user_sub_update_user_agent(access_token):
         cleanup_groups(access_token, core, groups)
 
 
+def test_user_sub_update_user_agent_truncates_long_values(access_token):
+    """Ensure overly long User-Agent strings are stored without failing."""
+    core, groups = setup_groups(access_token, 1)
+    user = create_user(
+        access_token,
+        group_ids=[groups[0]["id"]],
+        payload={"username": unique_name("test_user_agent_truncate")},
+    )
+    try:
+        url = user["subscription_url"]
+        long_user_agent = "A" * 1000
+        client.get(url, headers={"User-Agent": long_user_agent})
+        response = client.get(
+            f"/api/user/{user['username']}/sub_update",
+            headers={"Authorization": f"Bearer {access_token}"},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["updates"][0]["user_agent"] == long_user_agent[:512]
+    finally:
+        delete_user(access_token, user["username"])
+        cleanup_groups(access_token, core, groups)
+
+
 def test_user_get(access_token):
     """Test that the user get by id route is accessible."""
     core, groups = setup_groups(access_token, 1)
