T09 · Insecure Skill Coding Practices
Warning
- Location
- attio_client.py:214
- Finding
- Broken Pagination Causes Unbounded API Requests and Memory Consumption<![CDATA[ ## Vulnerability Details **File Location**: `attio_client.py:214-238` **Vulnerability Type**: Unbounded loop caused by ineffective pagination **Risk Level**: Medium ### Vulnerable Code ```python if args.all: # Auto-paginate to get all records all_records = [] batch_size = 1000 offset = 0 if endpoint in ["deals", "deal"]: fetch_func = list_deals elif endpoint in ["companies", "company"]: fetch_func = list_companies elif endpoint in ["people", "person", "contacts"]: fetch_func = list_people else: fetch_func = lambda lim: query_records(endpoint, limit=lim) while True: result = fetch_func(limit=batch_size) if "error" in result: print(json.dumps(result, indent=2)) sys.exit(1) records = result.get("data", []) all_records.extend(records) if len(records) < batch_size: break offset += batch_size print(f"Fetched {len(all_records)}...", file=sys.stderr) ``` ### Technical Analysis The `--all` implementation increments the local `offset` variable after each full page, but it never passes that offset to `fetch_func`. The selected helper functions only receive `limit=batch_size`, so every iteration requests the same first page. When the first response contains exactly 1,000 records, the termination condition remains false. The same records are consequently downloaded and appended to `all_records` indefinitely. The process continues issuing authenticated requests and consuming memory until it encounters an API error, is terminated, or exhausts available resources. This behavior also affects custom object endpoints because the fallback lambda accepts only a limit and calls `query_records` without an offset. ### Attack Path 1. An operator or automation process configures a valid `ATTIO_API_KEY`. 2. The relevant Attio object contains at least 1,000 records, causing the first ...[truncated 1490 chars]
- Remediation
- <![CDATA[ ## Remediation Suggestions Pass the current offset into every pagination request and impose explicit safety limits. Refactor the fetch functions to use a consistent signature, for example: ```python def list_companies(limit: int = 100, offset: int = 0): return query_records("companies", limit=limit, offset=offset) # Apply the same signature to list_deals and list_people. while True: result = fetch_func(limit=batch_size, offset=offset) if "error" in result: print(json.dumps(result, indent=2)) sys.exit(1) records = result.get("data", []) all_records.extend(records) if len(records) < batch_size: break offset += len(records) ``` For custom objects, preserve both arguments: ```python fetch_func = lambda limit, offset: query_records( endpoint, limit=limit, offset=offset, ) ``` Additional hardening should include: 1. Prefer pagination metadata or server-provided cursors over inferring completion solely from page length. 2. Add configurable maximum page and record counts to prevent accidental resource exhaustion. 3. Detect repeated page identifiers and abort if the server returns the same page. 4. Stream or incrementally process results instead of retaining every record in memory. 5. Return a nonzero exit status when pagination fails. 6. Add tests covering datasets with exactly 1,000 records, more than 1,000 records, repeated pages, empty pages, and API errors. ]]>
