Fix smart-flow resource routing for URLs with query strings

changelog.d/+smart-flow-url-paths.fixed.md

@@ -0,0 +1 @@
+Use only URL paths when extracting smart-flow resource IDs, so query strings and fragments do not split organization rate-limit buckets or corrupt network/device cache lookups.

meraki/smart_flow.py

@@ -19,6 +19,7 @@
 import time
 from datetime import datetime, timezone
 from pathlib import Path
+from urllib.parse import urlsplit
 from typing import Any, Callable, Coroutine, Dict, Optional, Set
 
 import json
@@ -199,15 +200,15 @@ def _get_or_create_bucket(self, org_id: str) -> TokenBucket:
 
     def resolve_org(self, url: str) -> Optional[str]:
         """Extract org ID from URL, using cache for network/device lookups."""
-        m = _ORG_PATTERN.search(url)
+        m = _ORG_PATTERN.search(urlsplit(url).path)
         if m:
             return m.group(1)
 
-        m = _NETWORK_PATTERN.search(url)
+        m = _NETWORK_PATTERN.search(urlsplit(url).path)
         if m:
             return self._network_to_org.get(m.group(1))
 
-        m = _DEVICE_PATTERN.search(url)
+        m = _DEVICE_PATTERN.search(urlsplit(url).path)
         if m:
             return self._serial_to_org.get(m.group(1))
 
@@ -231,11 +232,11 @@ def _resolve_inline(self, url: str) -> None:
         if not self._resolver:
             return
 
-        m = _NETWORK_PATTERN.search(url)
+        m = _NETWORK_PATTERN.search(urlsplit(url).path)
         if m:
             identifier, id_type = m.group(1), "network"
         else:
-            m = _DEVICE_PATTERN.search(url)
+            m = _DEVICE_PATTERN.search(urlsplit(url).path)
             if m:
                 identifier, id_type = m.group(1), "device"
             else:
@@ -377,17 +378,17 @@ def _learn_from_body(self, body: dict, org_id: str) -> tuple:
 
     @staticmethod
     def _org_id_from_url(url: str) -> Optional[str]:
-        m = _ORG_PATTERN.search(url)
+        m = _ORG_PATTERN.search(urlsplit(url).path)
         return m.group(1) if m else None
 
     @staticmethod
     def _network_id_from_url(url: str) -> Optional[str]:
-        m = _NETWORK_PATTERN.search(url)
+        m = _NETWORK_PATTERN.search(urlsplit(url).path)
         return m.group(1) if m else None
 
     @staticmethod
     def _serial_from_url(url: str) -> Optional[str]:
-        m = _DEVICE_PATTERN.search(url)
+        m = _DEVICE_PATTERN.search(urlsplit(url).path)
         return m.group(1) if m else None
 
     @staticmethod
@@ -529,15 +530,15 @@ def _get_or_create_bucket(self, org_id: str) -> AsyncTokenBucket:
 
     def resolve_org(self, url: str) -> Optional[str]:
         """Extract org ID from URL, using cache for network/device lookups."""
-        m = _ORG_PATTERN.search(url)
+        m = _ORG_PATTERN.search(urlsplit(url).path)
         if m:
             return m.group(1)
 
-        m = _NETWORK_PATTERN.search(url)
+        m = _NETWORK_PATTERN.search(urlsplit(url).path)
         if m:
             return self._network_to_org.get(m.group(1))
 
-        m = _DEVICE_PATTERN.search(url)
+        m = _DEVICE_PATTERN.search(urlsplit(url).path)
         if m:
             return self._serial_to_org.get(m.group(1))
 
@@ -558,11 +559,11 @@ def _trigger_background_resolve(self, url: str) -> None:
         if not self._resolver:
             return
 
-        m = _NETWORK_PATTERN.search(url)
+        m = _NETWORK_PATTERN.search(urlsplit(url).path)
         if m:
             identifier, id_type = m.group(1), "network"
         else:
-            m = _DEVICE_PATTERN.search(url)
+            m = _DEVICE_PATTERN.search(urlsplit(url).path)
             if m:
                 identifier, id_type = m.group(1), "device"
             else:

tests/unit/test_smart_flow_url_paths.py

@@ -0,0 +1,55 @@
+"""Resource routing must use URL paths rather than query or fragment text."""
+
+import asyncio
+
+import pytest
+
+from meraki.smart_flow import AsyncOrgRateLimiter, OrgRateLimiter
+
+
+@pytest.mark.parametrize("limiter_class", [OrgRateLimiter, AsyncOrgRateLimiter])
+@pytest.mark.parametrize("resource", ["organizations/org_1", "networks/N_1", "devices/Q123"])
+@pytest.mark.parametrize("suffix", ["?perPage=10", "#details", "?next=/organizations/other/networks"])
+def test_resolve_org_ignores_query_and_fragment(limiter_class, resource, suffix):
+    limiter = limiter_class()
+    limiter.register_network("N_1", "org_1")
+    limiter.register_device("Q123", "org_1")
+    assert limiter.resolve_org(f"https://api.meraki.com/api/v1/{resource}{suffix}") == "org_1"
+
+
+@pytest.mark.parametrize("limiter_class", [OrgRateLimiter, AsyncOrgRateLimiter])
+def test_query_paths_do_not_create_resource_matches(limiter_class):
+    limiter = limiter_class()
+    assert limiter.resolve_org("/admin?redirect=/organizations/org_1/networks") is None
+
+
+@pytest.mark.parametrize("resource, id_type, identifier", [("networks", "network", "N_1"), ("devices", "device", "Q123")])
+def test_sync_resolver_receives_only_resource_identifier(resource, id_type, identifier):
+    calls = []
+    limiter = OrgRateLimiter()
+    limiter.set_resolver(lambda kind, value: calls.append((kind, value)) or "org_1")
+    limiter.acquire(f"/{resource}/{identifier}?perPage=10")
+    assert calls == [(id_type, identifier)]
+    assert limiter.resolve_org(f"/{resource}/{identifier}") == "org_1"
+
+
+@pytest.mark.asyncio
+async def test_async_resolver_receives_only_resource_identifier():
+    calls = []
+
+    async def resolver(kind, value):
+        calls.append((kind, value))
+        return "org_1"
+
+    limiter = AsyncOrgRateLimiter()
+    limiter.set_resolver(resolver)
+    await limiter.acquire("/networks/N_1?perPage=10")
+    await asyncio.gather(*limiter._bg_tasks)
+    assert calls == [("network", "N_1")]
+    assert limiter.resolve_org("/networks/N_1") == "org_1"
+
+
+def test_response_learning_uses_path_identifiers():
+    limiter = OrgRateLimiter()
+    limiter.learn_from_response("/networks/N_1?perPage=10", {"organizationId": "org_1"})
+    assert limiter.resolve_org("/networks/N_1") == "org_1"