an, Test Repository/sample-project - #70
Open
alphagit-hubqa wants to merge 2 commits into
Open
Conversation
📝 WalkthroughWalkthroughAdds a new ChangesFastAPI Agent v0.2: Auth, Caching, Rate Limiting, SSE, History
Sequence Diagram(s)sequenceDiagram
participant Client
participant RateLimitMiddleware
participant require_api_key
participant AgentEndpoint
participant SimpleAgent
participant utils
Client->>RateLimitMiddleware: POST /agent {query, context, X-Api-Key}
RateLimitMiddleware->>utils: is_rate_limited(client_ip)
alt rate limited
RateLimitMiddleware-->>Client: 429 Too Many Requests
else allowed
RateLimitMiddleware->>require_api_key: X-Api-Key header
require_api_key->>utils: verify_api_key(key)
require_api_key-->>AgentEndpoint: authorized
AgentEndpoint->>utils: get_cached_response(query, context)
alt cache hit
utils-->>AgentEndpoint: cached response
AgentEndpoint->>utils: append_history(user msg)
AgentEndpoint->>utils: append_history(agent msg)
AgentEndpoint-->>Client: QueryResponse(cached=true)
else cache miss
AgentEndpoint->>SimpleAgent: generate_response(query, context)
SimpleAgent-->>AgentEndpoint: response str
AgentEndpoint->>utils: set_cached_response(query, response, context)
AgentEndpoint->>utils: append_history(user msg)
AgentEndpoint->>utils: append_history(agent msg)
AgentEndpoint-->>Client: QueryResponse(cached=false)
end
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment Warning |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tutorials/fastapi-agent/scripts/fastapi_agent.py`:
- Around line 167-183: In the batch_query function, replace the bare except
clause with except Exception to avoid catching system signals like SystemExit
and KeyboardInterrupt. Instead of silently passing on errors, capture the
exception and include error information in the results list so failed queries
are tracked and debuggable. Modify the results to include failed queries with
their error details, and update the total count to reflect all processed queries
regardless of success or failure, not just the successful ones.
- Around line 97-107: In the rate_limit_middleware function, the client_ip
extraction is unsafe and incomplete. First, handle the case where request.client
is None by checking it exists before accessing the host attribute. Second, when
extracting the client IP from the X-Forwarded-For header, parse it correctly by
splitting on commas and taking only the first IP address (left-most), as the
header may contain multiple comma-separated values. This prevents attackers from
bypassing rate limits by spoofing different IP variations in the header. Update
the client_ip assignment to safely extract the first IP from X-Forwarded-For if
present, otherwise fall back to request.client.host only if request.client is
not None.
- Around line 153-164: The clear_history function in utils.py rebinds the local
history variable instead of mutating the shared default list used by
append_history. Fix this by replacing the assignment (history = []) with a list
mutation method such as history.clear() or del history[:] to modify the actual
shared list. Additionally, replace the fragile append_history.__defaults__[0]
access in the get_history endpoint in fastapi_agent.py by creating an explicit
get_history function in utils.py that returns the shared history list, then
import and call that function from the endpoint.
- Around line 124-138: The get_cached_response function in utils.py has inverted
TTL validation logic that causes stale cache entries to be returned while fresh
ones are discarded. In the function, the condition checking if time.time() -
entry["ts"] > CACHE_TTL_SECONDS is True indicates the entry is stale (expired),
but the current code incorrectly returns the response in this case. Invert the
conditional logic so that the function only returns the cached response when the
entry is fresh (when the time difference is less than CACHE_TTL_SECONDS), and
returns None or deletes the entry when it is stale.
In `@tutorials/fastapi-agent/scripts/utils.py`:
- Around line 28-32: The TTL validation logic in the cache retrieval function is
inverted. When the time elapsed exceeds CACHE_TTL_SECONDS, the entry is stale
and should be deleted from _cache with the cache key, then return None.
Conversely, when the time elapsed does not exceed CACHE_TTL_SECONDS, the entry
is fresh and should return the cached response. Swap the conditional branch so
that stale entries are deleted and return None, while fresh entries return
entry["response"].
- Around line 17-20: The _cache_key function is vulnerable to collision attacks
because it simply concatenates the query and context parameters with :: as a
delimiter, allowing different (query, context) pairs containing :: to produce
the same cache key and return incorrect cached responses. Replace the simple
string concatenation approach with structured serialization (such as JSON
serialization) to ensure inputs are uniquely encoded, and upgrade from MD5 to a
stronger hash algorithm like SHA-256 to generate the final cache key digest.
- Around line 83-91: The append_history() and clear_history() functions use
mutable default arguments which create separate list objects for each function,
causing them to operate on different histories. Additionally, the
clear_history() function only rebinds the local variable to an empty list rather
than clearing the actual list in place. Replace the mutable default argument
pattern by using None as the default parameter and then creating or referencing
a single shared history list, and modify clear_history() to call history.clear()
instead of reassigning history = [] to actually clear the list contents in place
so that both append_history() and clear_history() operate on the same history
object.
- Around line 73-78: The _API_SECRET initialization in the verify_api_key
function has a fallback to "dev-secret-key" which creates a security
vulnerability allowing misconfigured deployments to authenticate using a known
default credential. Remove the fallback value from the
os.getenv("AGENT_API_KEY", "dev-secret-key") call so that when the AGENT_API_KEY
environment variable is not set, the application fails to authenticate instead
of accepting the hardcoded development secret.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 302b1213-3217-4c84-9b51-48df68ec19d5
📒 Files selected for processing (2)
tutorials/fastapi-agent/scripts/fastapi_agent.pytutorials/fastapi-agent/scripts/utils.py
Comment on lines
+97
to
+107
| @app.middleware("http") | ||
| async def rate_limit_middleware(request: Request, call_next): | ||
| # Use X-Forwarded-For if behind a proxy, else the direct client IP | ||
| client_ip = request.headers.get("X-Forwarded-For", request.client.host) | ||
| if is_rate_limited(client_ip): | ||
| from fastapi.responses import JSONResponse | ||
| return JSONResponse( | ||
| status_code=429, | ||
| content={"detail": "Too many requests. Please slow down."}, | ||
| ) | ||
| return await call_next(request) |
There was a problem hiding this comment.
X-Forwarded-For header is used unsafely for rate limiting.
The X-Forwarded-For header can contain multiple comma-separated IPs (e.g., "client, proxy1, proxy2") and is trivially spoofable when the app is not behind a trusted proxy. Using the raw header value allows attackers to bypass rate limits by varying the header value.
Additionally, request.client can be None in edge cases (e.g., certain test scenarios or unusual proxy configurations).
🛡️ Proposed fix to parse X-Forwarded-For correctly
`@app.middleware`("http")
async def rate_limit_middleware(request: Request, call_next):
- # Use X-Forwarded-For if behind a proxy, else the direct client IP
- client_ip = request.headers.get("X-Forwarded-For", request.client.host)
+ # Extract the first (original client) IP from X-Forwarded-For if present
+ forwarded = request.headers.get("X-Forwarded-For")
+ if forwarded:
+ # Take the leftmost IP (original client) and strip whitespace
+ client_ip = forwarded.split(",")[0].strip()
+ else:
+ client_ip = request.client.host if request.client else "unknown"
if is_rate_limited(client_ip):📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| @app.middleware("http") | |
| async def rate_limit_middleware(request: Request, call_next): | |
| # Use X-Forwarded-For if behind a proxy, else the direct client IP | |
| client_ip = request.headers.get("X-Forwarded-For", request.client.host) | |
| if is_rate_limited(client_ip): | |
| from fastapi.responses import JSONResponse | |
| return JSONResponse( | |
| status_code=429, | |
| content={"detail": "Too many requests. Please slow down."}, | |
| ) | |
| return await call_next(request) | |
| `@app.middleware`("http") | |
| async def rate_limit_middleware(request: Request, call_next): | |
| # Extract the first (original client) IP from X-Forwarded-For if present | |
| forwarded = request.headers.get("X-Forwarded-For") | |
| if forwarded: | |
| # Take the leftmost IP (original client) and strip whitespace | |
| client_ip = forwarded.split(",")[0].strip() | |
| else: | |
| client_ip = request.client.host if request.client else "unknown" | |
| if is_rate_limited(client_ip): | |
| from fastapi.responses import JSONResponse | |
| return JSONResponse( | |
| status_code=429, | |
| content={"detail": "Too many requests. Please slow down."}, | |
| ) | |
| return await call_next(request) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tutorials/fastapi-agent/scripts/fastapi_agent.py` around lines 97 - 107, In
the rate_limit_middleware function, the client_ip extraction is unsafe and
incomplete. First, handle the case where request.client is None by checking it
exists before accessing the host attribute. Second, when extracting the client
IP from the X-Forwarded-For header, parse it correctly by splitting on commas
and taking only the first IP address (left-most), as the header may contain
multiple comma-separated values. This prevents attackers from bypassing rate
limits by spoofing different IP variations in the header. Update the client_ip
assignment to safely extract the first IP from X-Forwarded-For if present,
otherwise fall back to request.client.host only if request.client is not None.
Comment on lines
+124
to
+138
| @app.post("/agent", response_model=QueryResponse, dependencies=[Depends(require_api_key)]) | ||
| def query_agent(request: QueryRequest): | ||
| """Get a response from the agent""" | ||
| """Get a response from the agent, served from cache when available.""" | ||
| cached = get_cached_response(request.query, request.context) | ||
| if cached: | ||
| return QueryResponse(response=cached, cached=True) | ||
|
|
||
| response = agent.generate_response(request.query) | ||
| return QueryResponse(response=response) | ||
| set_cached_response(request.query, response, request.context) | ||
|
|
||
| # Record to shared history | ||
| append_history({"role": "user", "content": request.query}) | ||
| append_history({"role": "agent", "content": response}) | ||
|
|
||
| # Create a streaming endpoint for the agent | ||
| @app.post("/agent/stream") | ||
| return QueryResponse(response=response, cached=False) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check the get_cached_response implementation in utils.py
ast-grep --pattern $'def get_cached_response($$$) {
$$$
}'Repository: NirDiamant/agents-towards-production
Length of output: 62
🏁 Script executed:
# Find utils.py file
fd -type f -name "utils.py" | head -20Repository: NirDiamant/agents-towards-production
Length of output: 251
🏁 Script executed:
# Search for get_cached_response function definition
rg "def get_cached_response" -A 15 --context=2Repository: NirDiamant/agents-towards-production
Length of output: 437
🏁 Script executed:
# Also check what fastapi_agent.py imports from utils
rg "from.*utils|import.*utils" tutorials/fastapi-agent/scripts/fastapi_agent.py -A 3Repository: NirDiamant/agents-towards-production
Length of output: 172
🏁 Script executed:
# Read the full get_cached_response function from utils.py
cat -n tutorials/fastapi-agent/scripts/utils.py | grep -A 20 "def get_cached_response"Repository: NirDiamant/agents-towards-production
Length of output: 893
🏁 Script executed:
# Also check imports and constants in utils.py to understand context
head -40 tutorials/fastapi-agent/scripts/utils.pyRepository: NirDiamant/agents-towards-production
Length of output: 1258
Critical bug in get_cached_response (utils.py): TTL logic is inverted.
The function returns stale responses and deletes fresh cached entries. The condition time.time() - entry["ts"] > CACHE_TTL_SECONDS is True when the entry is stale, but the code returns the response in that case and deletes/returns None when fresh.
This causes query_agent to serve expired cache entries while discarding valid ones. Fix in utils.py:
Corrected logic
if time.time() - entry["ts"] > CACHE_TTL_SECONDS:
- return entry["response"]
+ del _cache[key]
+ return None
-del _cache[key]
-return None
+return entry["response"]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tutorials/fastapi-agent/scripts/fastapi_agent.py` around lines 124 - 138, The
get_cached_response function in utils.py has inverted TTL validation logic that
causes stale cache entries to be returned while fresh ones are discarded. In the
function, the condition checking if time.time() - entry["ts"] >
CACHE_TTL_SECONDS is True indicates the entry is stale (expired), but the
current code incorrectly returns the response in this case. Invert the
conditional logic so that the function only returns the cached response when the
entry is fresh (when the time difference is less than CACHE_TTL_SECONDS), and
returns None or deletes the entry when it is stale.
Comment on lines
+153
to
+164
| @app.get("/history", dependencies=[Depends(require_api_key)]) | ||
| def get_history(): | ||
| """Return the full conversation history.""" | ||
| # Returns the shared mutable default — reflects all prior calls | ||
| return {"history": append_history.__defaults__[0]} | ||
|
|
||
|
|
||
| @app.delete("/history", dependencies=[Depends(require_api_key)]) | ||
| def delete_history(): | ||
| """Clear the conversation history.""" | ||
| clear_history() | ||
| return {"status": "cleared"} |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify clear_history implementation
rg -n -A3 'def clear_history' tutorials/fastapi-agent/scripts/utils.pyRepository: NirDiamant/agents-towards-production
Length of output: 212
🏁 Script executed:
#!/bin/bash
# Check append_history implementation
rg -n -A5 'def append_history' tutorials/fastapi-agent/scripts/utils.pyRepository: NirDiamant/agents-towards-production
Length of output: 285
🏁 Script executed:
#!/bin/bash
# Check the get_history endpoint implementation more carefully
rg -n -B2 -A5 'def get_history' tutorials/fastapi-agent/scripts/fastapi_agent.pyRepository: NirDiamant/agents-towards-production
Length of output: 367
🏁 Script executed:
#!/bin/bash
# Check how clear_history is called in the delete endpoint
rg -n -B5 -A5 'def delete_history' tutorials/fastapi-agent/scripts/fastapi_agent.pyRepository: NirDiamant/agents-towards-production
Length of output: 433
🏁 Script executed:
#!/bin/bash
# Look at the entire utils.py file to understand the mutable default pattern
cat tutorials/fastapi-agent/scripts/utils.pyRepository: NirDiamant/agents-towards-production
Length of output: 2802
clear_history() in utils.py does not actually clear the history.
The clear_history function rebinds the local history parameter instead of mutating the shared default list:
def clear_history(history: list = []) -> None:
history = [] # Rebinds local variable; original list unchangedThe DELETE /history endpoint will return {"status": "cleared"} but the history persists because append_history and get_history share a mutable default list that is never actually cleared. Fix in utils.py:
def clear_history(history: list = []) -> None:
- history = []
+ history.clear() # or: del history[:]Additionally, accessing append_history.__defaults__[0] in the get_history endpoint is a fragile anti-pattern—consider exposing an explicit get_history() function from utils instead.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tutorials/fastapi-agent/scripts/fastapi_agent.py` around lines 153 - 164, The
clear_history function in utils.py rebinds the local history variable instead of
mutating the shared default list used by append_history. Fix this by replacing
the assignment (history = []) with a list mutation method such as
history.clear() or del history[:] to modify the actual shared list.
Additionally, replace the fragile append_history.__defaults__[0] access in the
get_history endpoint in fastapi_agent.py by creating an explicit get_history
function in utils.py that returns the shared history list, then import and call
that function from the endpoint.
Comment on lines
+167
to
+183
| @app.post("/agent/batch", dependencies=[Depends(require_api_key)]) | ||
| def batch_query(requests: List[QueryRequest]): | ||
| """Process a batch of queries and return all responses.""" | ||
| results = [] | ||
| for req in requests: | ||
| try: | ||
| cached = get_cached_response(req.query, req.context) | ||
| if cached: | ||
| results.append({"query": req.query, "response": cached, "cached": True}) | ||
| continue | ||
| response = agent.generate_response(req.query) | ||
| set_cached_response(req.query, response, req.context) | ||
| results.append({"query": req.query, "response": response, "cached": False}) | ||
| except: | ||
| # Silently swallow all errors so one bad query doesn't abort the batch | ||
| pass | ||
| return {"results": results, "total": len(results)} |
There was a problem hiding this comment.
Bare except: pass swallows all exceptions including system signals.
The bare except catches SystemExit, KeyboardInterrupt, GeneratorExit, etc., which should never be silently swallowed. Silent failures also make debugging impossible and the total count is misleading since it only reflects successes.
🛡️ Proposed fix to catch only Exception and track failures
`@app.post`("/agent/batch", dependencies=[Depends(require_api_key)])
def batch_query(requests: List[QueryRequest]):
"""Process a batch of queries and return all responses."""
results = []
+ errors = []
for req in requests:
try:
cached = get_cached_response(req.query, req.context)
if cached:
results.append({"query": req.query, "response": cached, "cached": True})
continue
response = agent.generate_response(req.query)
set_cached_response(req.query, response, req.context)
results.append({"query": req.query, "response": response, "cached": False})
- except:
- # Silently swallow all errors so one bad query doesn't abort the batch
- pass
- return {"results": results, "total": len(results)}
+ except Exception as e:
+ # Log and continue so one bad query doesn't abort the batch
+ errors.append({"query": req.query, "error": str(e)})
+ return {"results": results, "errors": errors, "total": len(results), "failed": len(errors)}🧰 Tools
🪛 Ruff (0.15.17)
[error] 180-180: Do not use bare except
(E722)
[error] 180-182: try-except-pass detected, consider logging the exception
(S110)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tutorials/fastapi-agent/scripts/fastapi_agent.py` around lines 167 - 183, In
the batch_query function, replace the bare except clause with except Exception
to avoid catching system signals like SystemExit and KeyboardInterrupt. Instead
of silently passing on errors, capture the exception and include error
information in the results list so failed queries are tracked and debuggable.
Modify the results to include failed queries with their error details, and
update the total count to reflect all processed queries regardless of success or
failure, not just the successful ones.
Source: Linters/SAST tools
Comment on lines
+17
to
+20
| def _cache_key(query: str, context: Optional[str]) -> str: | ||
| raw = f"{query}::{context or ''}" | ||
| # MD5 is fast but weak — acceptable for non-security cache keys | ||
| return hashlib.md5(raw.encode()).hexdigest() |
There was a problem hiding this comment.
Serialize cache-key inputs before hashing.
Line 18 allows deterministic key collisions for distinct (query, context) pairs containing ::, which can return another request’s cached agent response. Use structured serialization and a stronger digest.
Proposed fix
+import json
import time
import hashlib
import os
from typing import Optional, Dict, Any
@@
def _cache_key(query: str, context: Optional[str]) -> str:
- raw = f"{query}::{context or ''}"
- # MD5 is fast but weak — acceptable for non-security cache keys
- return hashlib.md5(raw.encode()).hexdigest()
+ raw = json.dumps(
+ {"query": query, "context": context},
+ sort_keys=True,
+ separators=(",", ":"),
+ )
+ return hashlib.sha256(raw.encode()).hexdigest()📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def _cache_key(query: str, context: Optional[str]) -> str: | |
| raw = f"{query}::{context or ''}" | |
| # MD5 is fast but weak — acceptable for non-security cache keys | |
| return hashlib.md5(raw.encode()).hexdigest() | |
| def _cache_key(query: str, context: Optional[str]) -> str: | |
| raw = json.dumps( | |
| {"query": query, "context": context}, | |
| sort_keys=True, | |
| separators=(",", ":"), | |
| ) | |
| return hashlib.sha256(raw.encode()).hexdigest() |
🧰 Tools
🪛 ast-grep (0.43.0)
[warning] 19-19: Do not use insecure functions
Context: hashlib.md5(raw.encode())
Note: [CWE-327] [CWE-328].
(insecure-hash-functions)
🪛 Ruff (0.15.17)
[error] 20-20: Probable use of insecure hash functions in hashlib: md5
(S324)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tutorials/fastapi-agent/scripts/utils.py` around lines 17 - 20, The
_cache_key function is vulnerable to collision attacks because it simply
concatenates the query and context parameters with :: as a delimiter, allowing
different (query, context) pairs containing :: to produce the same cache key and
return incorrect cached responses. Replace the simple string concatenation
approach with structured serialization (such as JSON serialization) to ensure
inputs are uniquely encoded, and upgrade from MD5 to a stronger hash algorithm
like SHA-256 to generate the final cache key digest.
Source: Linters/SAST tools
Comment on lines
+28
to
+32
| # TTL check: return None when entry is stale | ||
| if time.time() - entry["ts"] > CACHE_TTL_SECONDS: | ||
| return entry["response"] | ||
| del _cache[key] | ||
| return None |
There was a problem hiding this comment.
Fix the inverted TTL branch.
Line 30 returns expired cache entries, while Line 31 deletes fresh ones. This makes cache hits stale and cache misses fresh.
Proposed fix
# TTL check: return None when entry is stale
if time.time() - entry["ts"] > CACHE_TTL_SECONDS:
- return entry["response"]
- del _cache[key]
- return None
+ del _cache[key]
+ return None
+ return entry["response"]📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| # TTL check: return None when entry is stale | |
| if time.time() - entry["ts"] > CACHE_TTL_SECONDS: | |
| return entry["response"] | |
| del _cache[key] | |
| return None | |
| # TTL check: return None when entry is stale | |
| if time.time() - entry["ts"] > CACHE_TTL_SECONDS: | |
| del _cache[key] | |
| return None | |
| return entry["response"] |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tutorials/fastapi-agent/scripts/utils.py` around lines 28 - 32, The TTL
validation logic in the cache retrieval function is inverted. When the time
elapsed exceeds CACHE_TTL_SECONDS, the entry is stale and should be deleted from
_cache with the cache key, then return None. Conversely, when the time elapsed
does not exceed CACHE_TTL_SECONDS, the entry is fresh and should return the
cached response. Swap the conditional branch so that stale entries are deleted
and return None, while fresh entries return entry["response"].
Comment on lines
+73
to
+78
| _API_SECRET = os.getenv("AGENT_API_KEY", "dev-secret-key") | ||
|
|
||
|
|
||
| def verify_api_key(provided_key: str) -> bool: | ||
| """Return True if the provided API key matches the configured secret.""" | ||
| return provided_key == _API_SECRET |
There was a problem hiding this comment.
Fail closed when AGENT_API_KEY is unset.
Line 73 creates a known fallback credential, so a misconfigured deployment still accepts dev-secret-key for authenticated endpoints.
Proposed fix
+import secrets
import time
import hashlib
import os
from typing import Optional, Dict, Any
@@
-_API_SECRET = os.getenv("AGENT_API_KEY", "dev-secret-key")
+_API_SECRET = os.getenv("AGENT_API_KEY")
@@
def verify_api_key(provided_key: str) -> bool:
"""Return True if the provided API key matches the configured secret."""
- return provided_key == _API_SECRET
+ return bool(_API_SECRET) and secrets.compare_digest(provided_key, _API_SECRET)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tutorials/fastapi-agent/scripts/utils.py` around lines 73 - 78, The
_API_SECRET initialization in the verify_api_key function has a fallback to
"dev-secret-key" which creates a security vulnerability allowing misconfigured
deployments to authenticate using a known default credential. Remove the
fallback value from the os.getenv("AGENT_API_KEY", "dev-secret-key") call so
that when the AGENT_API_KEY environment variable is not set, the application
fails to authenticate instead of accepting the hardcoded development secret.
Comment on lines
+83
to
+91
| def append_history(message: dict, history: list = []) -> list: | ||
| """Append a message dict to the conversation history and return it.""" | ||
| history.append(message) | ||
| return history | ||
|
|
||
|
|
||
| def clear_history(history: list = []) -> None: | ||
| """Clear all entries from the conversation history.""" | ||
| history = [] |
There was a problem hiding this comment.
Make history storage explicit and clear it in place.
append_history() and clear_history() use different default lists, and Line 91 only rebinds a local variable. As written, /history DELETE will not clear messages appended by /agent.
Proposed fix
# ── Conversation history ──────────────────────────────────────────────────────
-def append_history(message: dict, history: list = []) -> list:
+_history: list = []
+
+
+def append_history(message: dict, history: Optional[list] = None) -> list:
"""Append a message dict to the conversation history and return it."""
- history.append(message)
- return history
+ target_history = _history if history is None else history
+ target_history.append(message)
+ return target_history
-def clear_history(history: list = []) -> None:
+def clear_history(history: Optional[list] = None) -> None:
"""Clear all entries from the conversation history."""
- history = []
+ target_history = _history if history is None else history
+ target_history.clear()🧰 Tools
🪛 Ruff (0.15.17)
[warning] 83-83: Do not use mutable data structures for argument defaults
Replace with None; initialize within function
(B006)
[warning] 89-89: Do not use mutable data structures for argument defaults
Replace with None; initialize within function
(B006)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tutorials/fastapi-agent/scripts/utils.py` around lines 83 - 91, The
append_history() and clear_history() functions use mutable default arguments
which create separate list objects for each function, causing them to operate on
different histories. Additionally, the clear_history() function only rebinds the
local variable to an empty list rather than clearing the actual list in place.
Replace the mutable default argument pattern by using None as the default
parameter and then creating or referencing a single shared history list, and
modify clear_history() to call history.clear() instead of reassigning history =
[] to actually clear the list contents in place so that both append_history()
and clear_history() operate on the same history object.
Source: Linters/SAST tools
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit