Feature: One whitelist IP per Token (Device Tokens) #23
Conversation
- Update core.py to support token_id in whitelist entries and enforce uniqueness - Update main.py to generate token_id from API key hash - Add tests covering new behavior and legacy fallback - Resolves #21
WalkthroughImplements one-IP-per-token device tokens and mixed-format whitelist support: whitelist entries may be legacy ints or dicts with expiry and Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Client
participant Main as src/main.py
participant Core as src/core.py
participant FS as Filesystem
participant FW as Firewalld
Client->>Main: Knock request (api_key, ip)
Main->>Main: token_id = SHA256(api_key) (if api_key)
Main->>Core: add_ip_to_whitelist_with_firewalld(ip, expiry, settings, token_id)
Core->>FS: load_whitelist()
FS-->>Core: whitelist (mixed legacy/new entries)
alt token_id present and maps to old_ip
Core->>FW: remove firewall rule for old_ip
FW-->>Core: removed / error
Core->>Core: record removed IP
else no replacement
Core->>Core: no prior owner found
end
Core->>Core: insert/update entry for ip with expiry and token_id
Core->>FW: add firewall rule for new ip
FW-->>Core: added / error
Core->>FS: save_whitelist(atomic)
FS-->>Core: persisted
Core-->>Main: return removed IP (or None)
Main-->>Client: response (success / error)
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes
Possibly related PRs
Suggested labels
Poem
Pre-merge checks and finishing touches✅ Passed checks (5 passed)
✨ Finishing touches🧪 Generate unit tests (beta)
📜 Recent review detailsConfiguration used: CodeRabbit UI Review profile: CHILL Plan: Pro 📒 Files selected for processing (3)
🧰 Additional context used🧠 Learnings (2)📚 Learning: 2025-09-08T22:31:37.677ZApplied to files:
📚 Learning: 2025-09-08T22:31:37.677ZApplied to files:
🔇 Additional comments (2)
Tip 📝 Customizable high-level summaries are now available in beta!You can now customize how CodeRabbit generates the high-level summary in your pull requests — including its content, structure, tone, and formatting.
Example instruction:
Note: This feature is currently in beta for Pro-tier users, and pricing will be announced later. 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 |
|
gemini 3 let's gooooooooooooooooooooooooooooooo |
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
|
|
||
|
|
||
| # Generate token_id from API key to enforce one-IP-per-token policy | ||
| token_id = hashlib.sha256(api_key.encode()).hexdigest() if api_key else None |
Check failure
Code scanning / CodeQL
Use of a broken or weak cryptographic hashing algorithm on sensitive data High
Show autofix suggestion
Hide autofix suggestion
Copilot Autofix
AI about 2 months ago
To address this issue, we should replace the use of hashlib.sha256 with a modern, computationally expensive key derivation function when hashing API keys, even for identification purposes. This ensures resistance to brute-force and pre-image attacks, aligning with best security practices for handling sensitive keys. The recommended approach is to use a KDF such as PBKDF2 from the built-in hashlib module (hashlib.pbkdf2_hmac) because it is readily available, widely accepted, and does not require third-party packages. For enhanced security, a unique per-application salt should be used. This requires defining a salt value, updating the hashing logic, and handling byte output as hexadecimal for consistency.
The changes are confined to the region where token_id is computed (line 510). We will introduce a static salt as a module-level variable (best stored securely or derived from configuration), implement the PBKDF2-based hash, and update token_id accordingly. We only need to add the definition for the salt and switch hashlib.sha256(...).hexdigest() to hashlib.pbkdf2_hmac(...).
| @@ -507,7 +507,14 @@ | ||
| # Use SHA256 to generate a deterministic ID for the token. | ||
| # This is NOT for password hashing, but for unique identification of the token | ||
| # to enforce the one-IP-per-token policy. The token itself is the secret. | ||
| token_id = hashlib.sha256(api_key.encode()).hexdigest() if api_key else None | ||
| token_id = ( | ||
| hashlib.pbkdf2_hmac( | ||
| "sha256", | ||
| api_key.encode(), | ||
| TOKEN_ID_SALT, | ||
| 100_000, | ||
| ).hex() if api_key else None | ||
| ) | ||
|
|
||
| # Add to whitelist with firewalld integration | ||
| # This will add firewalld rules BEFORE updating whitelist.json if firewalld is enabled |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/core.py (1)
198-217: Add storage_path validation to prevent path injection attacks.The config file path is properly validated in
load_config()(lines 40-49 of src/config.py), but thestorage_pathvalue extracted from settings is never validated. This allows directory traversal via payloads likestorage_path: "../../../etc/passwd"in the config YAML.Add validation in
load_config()after parsing the config (before line 84) to sanitize the whitelist storage_path:
- Resolve the path using
Path.resolve()- Validate it doesn't escape a safe base directory (e.g., application root)
- Use
Path.is_relative_to()to confirm containment (Python 3.9+)Alternatively, validate at the point of use in
load_whitelist()andsave_whitelist()(src/core.py lines 201, 224).
🧹 Nitpick comments (2)
src/core.py (2)
374-387: Consider using logging.exception for better debugging.The error handling correctly removes the old firewall rule when an IP is replaced. However, using
logging.errorinstead oflogging.exceptionloses the stack trace.Apply this diff to improve error logging:
except Exception as e: - logging.error(f"Failed to remove old firewall rule for {old_ip}: {e}") + logging.exception(f"Failed to remove old firewall rule for {old_ip}: {e}")As per static analysis hints.
389-403: Consider using logging.exception for better debugging.Similar to the previous comment, the rollback error handlers would benefit from stack traces for troubleshooting failed rollbacks.
Apply this diff:
try: firewalld_integration.remove_whitelist_rule(ip_or_cidr) - logging.error( + logging.exception( f"Rolled back firewalld rules for {ip_or_cidr} due to whitelist persistence failure: {e}" ) except Exception as rollback_error: - logging.error( + logging.exception( f"Failed to rollback firewalld rules for {ip_or_cidr}: {rollback_error}" ) - logging.error(f"Failed to persist whitelist entry for {ip_or_cidr}: {e}") + logging.exception(f"Failed to persist whitelist entry for {ip_or_cidr}: {e}")As per static analysis hints.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
src/core.py(10 hunks)src/main.py(20 hunks)tests/test_device_tokens.py(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (3)
src/main.py (3)
src/core.py (10)
cleanup_expired_ips(406-437)load_whitelist(198-216)is_valid_api_key(459-489)can_whitelist_remote(443-448)is_valid_ip_or_cidr(48-54)is_safe_cidr_range(57-78)get_max_ttl_for_key(451-456)add_ip_to_whitelist_with_firewalld(335-403)save_whitelist(219-259)is_ip_whitelisted(108-147)src/firewalld.py (4)
initialize_firewalld(591-595)is_enabled(78-80)setup_knocker_zone(222-306)restore_missing_rules(515-565)src/models.py (4)
KnockResponse(41-54)ErrorResponse(65-70)KnockRequest(9-38)HealthResponse(57-62)
tests/test_device_tokens.py (1)
src/core.py (2)
add_ip_to_whitelist(262-332)load_whitelist(198-216)
src/core.py (1)
src/firewalld.py (4)
get_firewalld_integration(586-588)is_enabled(78-80)add_whitelist_rule(357-407)remove_whitelist_rule(419-461)
🪛 GitHub Check: CodeQL
src/main.py
[failure] 507-507: Use of a broken or weak cryptographic hashing algorithm on sensitive data
Sensitive data (password) is used in a hashing algorithm (SHA256) that is insecure for password hashing, since it is not a computationally expensive hash function.
[failure] 536-536: Clear-text logging of sensitive information
This expression logs sensitive data (password) as clear text.
This expression logs sensitive data (password) as clear text.
src/core.py
[failure] 37-37: Uncontrolled data used in path expression
This path depends on a user-provided value.
This path depends on a user-provided value.
[failure] 202-202: Uncontrolled data used in path expression
This path depends on a user-provided value.
This path depends on a user-provided value.
[failure] 206-206: Uncontrolled data used in path expression
This path depends on a user-provided value.
This path depends on a user-provided value.
This path depends on a user-provided value.
[failure] 245-245: Uncontrolled data used in path expression
This path depends on a user-provided value.
This path depends on a user-provided value.
This path depends on a user-provided value.
🪛 Ruff (0.14.5)
src/main.py
170-170: Do not catch blind exception: Exception
(BLE001)
178-178: Do not catch blind exception: Exception
(BLE001)
179-181: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
300-300: Do not perform function call Depends in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
370-370: Do not perform function call Depends in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
589-591: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
602-602: Do not catch blind exception: Exception
(BLE001)
603-605: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
613-613: Do not catch blind exception: Exception
(BLE001)
614-614: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
624-624: Do not catch blind exception: Exception
(BLE001)
625-625: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
657-657: Do not perform function call Depends in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
tests/test_device_tokens.py
49-49: Possible hardcoded password assigned to: "token_id"
(S105)
75-75: Possible hardcoded password assigned to: "other_token"
(S105)
src/core.py
289-289: Avoid specifying long messages outside the exception class
(TRY003)
294-296: Avoid specifying long messages outside the exception class
(TRY003)
385-385: Do not catch blind exception: Exception
(BLE001)
386-386: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
388-388: Consider moving this statement to an else block
(TRY300)
389-389: Do not catch blind exception: Exception
(BLE001)
394-396: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
397-397: Do not catch blind exception: Exception
(BLE001)
398-400: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
402-402: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: build
🔇 Additional comments (11)
src/core.py (6)
108-147: LGTM! Enhanced whitelist checking with dict support.The updated logic correctly handles both legacy integer expiry values and new dict-based entries with token_id. The type checking with
isinstance(value, dict)ensures backward compatibility.
219-260: Verify path injection risk is mitigated.CodeQL flags line 245 for potential path injection. Similar to the load_whitelist function, the temp file path is derived from user-controlled settings without runtime validation.
Same verification as previous comments - check if config loading validates paths properly.
229-238: LGTM! Smart sorting helper for mixed value types.The
get_expiryhelper function correctly extracts expiry times from both dict and int entries for proper sorting when the whitelist exceeds size limits.
262-332: LGTM! Token-based IP replacement logic is correct.The implementation properly enforces one-IP-per-token:
- Lines 309-322: Correctly finds and removes existing IPs for the same token_id
- Lines 325-329: Properly stores new entries with token_id or as legacy int for backward compatibility
- Line 332: Returns the replaced IP for firewalld cleanup
The logic handles the edge case where the same IP is re-whitelisted (just updates TTL without replacement).
424-437: LGTM! Cleanup properly handles mixed value types.The cleanup logic correctly extracts expiry times from both dict and int entries, preserving the original value format when building the fresh whitelist.
459-489: LGTM! Constant-time comparison prevents timing attacks.The implementation correctly uses bitwise OR to prevent short-circuiting and iterates through all keys to maintain constant-time behavior. This is a security best practice for API key validation.
tests/test_device_tokens.py (1)
1-81: LGTM! Comprehensive test coverage for device token feature.The tests properly verify:
- Legacy behavior (lines 27-41): Multiple IPs allowed without token_id
- Per-token replacement (lines 44-72): Old IP removed when new IP added for same token_id, correct return value
- Cross-token isolation (lines 74-81): Different tokens don't interfere with each other
The fixture properly manages temporary whitelist files. Assertions are clear and comprehensive.
src/main.py (4)
507-507: SHA-256 usage is appropriate for token identification.CodeQL flags this as weak cryptography for password hashing, but this is a false positive. The code is not storing passwords—it's generating a deterministic identifier from the API key to enforce the one-IP-per-token policy. SHA-256 is appropriate here since:
- It's used for identification, not authentication
- Deterministic output is required (same token always generates same ID)
- The API keys themselves are already secrets stored securely
533-540: Review logging of potentially sensitive information.CodeQL flags line 536 for logging sensitive data. The code logs
ip_to_whitelistat DEBUG level. While IP addresses are generally not considered sensitive, in some contexts they could reveal user identity. The current DEBUG-level logging is a reasonable balance, but consider:
- The comment mentions avoiding logging at INFO level, which is good
ip_to_whitelistmight be sensitive depending on deployment context- Consider whether even DEBUG-level logging is necessary in production
Based on learnings or coding guidelines, is IP address logging at DEBUG level acceptable for this project's security requirements?
511-523: LGTM! Token ID properly passed to firewalld integration.The implementation correctly:
- Generates token_id from the API key (line 507)
- Passes it to
add_ip_to_whitelist_with_firewalld(line 512)- Enforces the one-IP-per-token policy at the core level
300-301: False positive: Depends() in defaults is FastAPI pattern.Ruff flags B008 here, but using
Depends()in function parameter defaults is the standard FastAPI dependency injection pattern. This is not an issue.
- Validate storage_path using .resolve() to prevent directory traversal - Redact API keys in logs (src/core.py: get_api_key_name) - Use logging.exception for better error tracing - Clarify SHA256 usage for token_id (not for auth)
| path = Path( | ||
| settings.get("whitelist", {}).get("storage_path", "whitelist.json") | ||
| ).resolve() |
Check failure
Code scanning / CodeQL
Uncontrolled data used in path expression High
Show autofix suggestion
Hide autofix suggestion
Copilot Autofix
AI about 2 months ago
To mitigate uncontrolled path usage, we should ensure that settings["whitelist"]["storage_path"] always resolves to a path within an intended “safe” directory—by default, the directory containing the application or a specific subdirectory, such as /var/app_data or similar. The resolution is best achieved by always joining the (possibly user-supplied) filename to a trusted base directory, normalizing the resulting path, and verifying that the final resolved path starts with the base directory path. If not, we should raise an exception or log and return an error, refusing to proceed. This fix involves defining a safe base directory (e.g., using a constant, or a setting initialized at startup and not modifiable by external users), updating the logic in load_whitelist() (and everywhere else that processes or writes to the whitelist file) to check/normalize the resolved file path, and enforcing that the whitelist is always read and written only within this directory. Additional imports are not strictly required, as pathlib is already imported, but we will need to insert code blocks for validation and error handling just prior to accessing the derived file.
| @@ -6,10 +6,10 @@ | ||
| import logging | ||
| import hmac | ||
| from pathlib import Path | ||
| import os | ||
| from typing import Dict, Any, Optional, Union | ||
| from contextlib import contextmanager | ||
|
|
||
| # Thread lock for whitelist operations | ||
| # Using RLock (reentrant lock) to allow nested lock acquisition | ||
| # in read-modify-write sequences | ||
| _whitelist_lock = threading.RLock() | ||
| @@ -198,10 +195,15 @@ | ||
| def load_whitelist(settings: Dict[str, Any]) -> Dict[str, Union[int, Dict[str, Any]]]: | ||
| """Loads the whitelist from the JSON file with thread safety.""" | ||
| with _whitelist_lock: | ||
| path = Path( | ||
| settings.get("whitelist", {}).get("storage_path", "whitelist.json") | ||
| ).resolve() | ||
| if not path.exists(): | ||
| # Always join to SAFE_WHITELIST_DIR and resolve, to prevent directory traversal | ||
| raw_storage_path = settings.get("whitelist", {}).get("storage_path", "whitelist.json") | ||
| storage_path = Path(raw_storage_path) | ||
| # Prevent absolute paths or traversal by always joining to the safe base | ||
| full_path = (SAFE_WHITELIST_DIR / storage_path).resolve() | ||
| # Ensure the resulting path is within the safe base directory | ||
| if not str(full_path).startswith(str(SAFE_WHITELIST_DIR) + os.sep): | ||
| raise Exception("Whitelist storage_path must be inside the designated directory.") | ||
| if not full_path.exists(): | ||
| return {} | ||
|
|
||
| try: |
| path = Path( | ||
| settings.get("whitelist", {}).get("storage_path", "whitelist.json") | ||
| ).resolve() |
Check failure
Code scanning / CodeQL
Uncontrolled data used in path expression High
Show autofix suggestion
Hide autofix suggestion
Copilot Autofix
AI about 2 months ago
To fix this issue, we need to ensure that any path derived from user input is validated before being used in file I/O operations, in particular for both saving and loading the whitelist. The recommended practice, especially if the path may span nested subfolders, is to resolve the computed path and ensure it remains inside a designated safe directory (base_path). We should:
- Define a root directory (
base_path) for whitelist storage (e.g.,/server/static/whitelist, or configured similarly). - After joining
base_pathwith the user-provided relative path, resolve and normalize the result. - Ensure that the resolved path starts with the
base_path(string or Path prefix). - Refuse operations (raise or fallback to default) if the path is outside the safe root.
Steps in code:
- In both
load_whitelistandsave_whitelist, compute the path asPath(base_path) / storage_path, resolve it, and check it is beneathbase_path. - If not, raise an exception or fallback to default safe file.
- Optionally define
base_pathvia settings with a secure default (never allow empty or/).
Regions to change:
- Inside
load_whitelistandsave_whitelistfunctions in src/core.py.
Imports:
No new external libraries are needed. Only standard Python (pathlib).
| @@ -196,11 +196,17 @@ | ||
|
|
||
|
|
||
| def load_whitelist(settings: Dict[str, Any]) -> Dict[str, Union[int, Dict[str, Any]]]: | ||
| """Loads the whitelist from the JSON file with thread safety.""" | ||
| """Loads the whitelist from the JSON file with thread safety and path validation.""" | ||
| with _whitelist_lock: | ||
| path = Path( | ||
| settings.get("whitelist", {}).get("storage_path", "whitelist.json") | ||
| ).resolve() | ||
| # Get base path and user path | ||
| base_path = Path(settings.get("whitelist", {}).get("storage_root", "whitelists/")).resolve() | ||
| storage_name = settings.get("whitelist", {}).get("storage_path", "whitelist.json") | ||
| # Compute the file path | ||
| path = (base_path / storage_name).resolve() | ||
| # Check that path is inside base_path | ||
| if not str(path).startswith(str(base_path)): | ||
| logging.error(f"Whitelist file path '{path}' is outside base directory '{base_path}'") | ||
| return {} | ||
| if not path.exists(): | ||
| return {} | ||
|
|
||
| @@ -221,11 +226,14 @@ | ||
| def save_whitelist( | ||
| whitelist: Dict[str, Union[int, Dict[str, Any]]], settings: Dict[str, Any] | ||
| ): | ||
| """Saves the whitelist to the JSON file with thread safety and size limits.""" | ||
| """Saves the whitelist to the JSON file with thread safety, size limits, and path validation.""" | ||
| with _whitelist_lock: | ||
| path = Path( | ||
| settings.get("whitelist", {}).get("storage_path", "whitelist.json") | ||
| ).resolve() | ||
| base_path = Path(settings.get("whitelist", {}).get("storage_root", "whitelists/")).resolve() | ||
| storage_name = settings.get("whitelist", {}).get("storage_path", "whitelist.json") | ||
| path = (base_path / storage_name).resolve() | ||
| if not str(path).startswith(str(base_path)): | ||
| logging.error(f"Whitelist file path '{path}' is outside base directory '{base_path}'") | ||
| raise Exception("Whitelist file path is outside the allowed directory") | ||
|
|
||
| # Security check: limit whitelist size to prevent DoS | ||
| max_entries = settings.get("security", {}).get("max_whitelist_entries", 10000) |
| settings.get("whitelist", {}).get("storage_path", "whitelist.json") | ||
| ).resolve() | ||
|
|
Check failure
Code scanning / CodeQL
Uncontrolled data used in path expression High
Show autofix suggestion
Hide autofix suggestion
Copilot Autofix
AI about 2 months ago
To fix this issue, we must ensure that the user-controlled value used to construct the whitelist file path is strictly contained within a safe, predefined directory and cannot escape via path traversal (e.g., ../), absolute paths, or symlinks. The "safe root" technique from the background is most appropriate here. In particular:
- Determine a constant, safe directory where whitelist storage files must live (e.g.,
/var/app/whitelistsor just the directory of the running application). This directory root should be hardcoded. - When constructing the file path from
settings.get("whitelist", {}).get("storage_path", ...), normalize it (usingos.path.normpathor equivalent), join it to the safe root, resolve it, and check that it starts with the safe root dir after resolution. - If the check fails, raise an exception (or fallback to a default file, or deny the request).
The code to update is the block at line 303–305 in add_ip_to_whitelist (in src/core.py).
You will also need to add an import for os (if not present) for path normalization, and define a constant safe root folder (e.g., WHITELIST_ROOT = Path(__file__).parent.resolve() or a fixed path).
Changes required:
- Add
WHITELIST_ROOT(as a module-level constant). - Validate the constructed path before use, ensuring containment inside
WHITELIST_ROOT. - Import
osif needed.
| @@ -6,9 +6,14 @@ | ||
| import logging | ||
| import hmac | ||
| from pathlib import Path | ||
| import os | ||
| from typing import Dict, Any, Optional, Union | ||
| from contextlib import contextmanager | ||
|
|
||
| # Root folder for all whitelist storage. | ||
| # You may change this to an absolute path if you want e.g. "/var/app/whitelists" | ||
| WHITELIST_ROOT = Path(__file__).parent.resolve() | ||
|
|
||
| # Thread lock for whitelist operations | ||
| # Using RLock (reentrant lock) to allow nested lock acquisition | ||
| # in read-modify-write sequences | ||
| @@ -300,9 +302,13 @@ | ||
| ) | ||
|
|
||
| # Get whitelist path for cross-process locking | ||
| whitelist_path = Path( | ||
| settings.get("whitelist", {}).get("storage_path", "whitelist.json") | ||
| ).resolve() | ||
| raw_storage_path = settings.get("whitelist", {}).get("storage_path", "whitelist.json") | ||
| # Prevent path traversal: only allow files within WHITELIST_ROOT | ||
| candidate_path = WHITELIST_ROOT / raw_storage_path | ||
| whitelist_path = candidate_path.resolve() | ||
| # Validate that the resolved path is inside WHITELIST_ROOT | ||
| if not str(whitelist_path).startswith(str(WHITELIST_ROOT) + os.sep): | ||
| raise ValueError(f"Whitelist storage path resolves outside allowed directory: {whitelist_path}") | ||
|
|
||
| # Use both in-process and cross-process locks | ||
| with _whitelist_lock: |
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
src/main.py (1)
593-593: Remove redundant exception objects from logging.exception calls.The
logging.exception()method automatically includes exception information in the traceback. Passing the exception object in the message string (e.g.,f"... {e}") is redundant and can make logs less clean.Apply this pattern to clean up the logging calls:
-logging.exception(f"Health check failed: Could not read whitelist storage: {e}") +logging.exception("Health check failed: Could not read whitelist storage")Similar changes should be applied to lines 607, 618, and 630.
As per static analysis hints.
Also applies to: 607-607, 618-618, 630-630
src/core.py (1)
391-391: Remove redundant exception objects from logging.exception calls.Similar to src/main.py, these
logging.exception()calls include redundant exception objects in the message string. The exception information is automatically included in the traceback.Apply this pattern:
-logging.exception(f"Failed to remove old firewall rule for {old_ip}: {e}") +logging.exception(f"Failed to remove old firewall rule for {old_ip}")Similar cleanup should be applied to lines 401, 405, and 408.
As per static analysis hints.
Also applies to: 401-401, 405-405, 408-408
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
src/core.py(10 hunks)src/main.py(20 hunks)
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2025-09-08T22:31:37.677Z
Learnt from: CR
Repo: FarisZR/knocker PR: 0
File: AGENTS.md:0-0
Timestamp: 2025-09-08T22:31:37.677Z
Learning: Applies to knocker.yaml : Set the whitelist storage path in knocker.yaml (JSON at /data/whitelist.json in container)
Applied to files:
src/core.py
🧬 Code graph analysis (2)
src/main.py (2)
src/core.py (10)
cleanup_expired_ips(412-443)load_whitelist(198-218)is_valid_api_key(465-495)can_whitelist_remote(449-454)is_valid_ip_or_cidr(48-54)is_safe_cidr_range(57-78)get_max_ttl_for_key(457-462)add_ip_to_whitelist_with_firewalld(339-409)save_whitelist(221-263)is_ip_whitelisted(108-147)src/firewalld.py (4)
initialize_firewalld(591-595)is_enabled(78-80)setup_knocker_zone(222-306)restore_missing_rules(515-565)
src/core.py (1)
src/firewalld.py (4)
get_firewalld_integration(586-588)is_enabled(78-80)add_whitelist_rule(357-407)remove_whitelist_rule(419-461)
🪛 GitHub Check: CodeQL
src/main.py
[failure] 510-510: Use of a broken or weak cryptographic hashing algorithm on sensitive data
Sensitive data (password) is used in a hashing algorithm (SHA256) that is insecure for password hashing, since it is not a computationally expensive hash function.
src/core.py
[failure] 201-203: Uncontrolled data used in path expression
This path depends on a user-provided value.
This path depends on a user-provided value.
This path depends on a user-provided value.
[failure] 204-204: Uncontrolled data used in path expression
This path depends on a user-provided value.
This path depends on a user-provided value.
[failure] 226-228: Uncontrolled data used in path expression
This path depends on a user-provided value.
This path depends on a user-provided value.
This path depends on a user-provided value.
[failure] 303-305: Uncontrolled data used in path expression
This path depends on a user-provided value.
[failure] 420-422: Uncontrolled data used in path expression
This path depends on a user-provided value.
🪛 Ruff (0.14.5)
src/main.py
170-170: Do not catch blind exception: Exception
(BLE001)
178-178: Do not catch blind exception: Exception
(BLE001)
179-181: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
300-300: Do not perform function call Depends in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
370-370: Do not perform function call Depends in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
593-593: Redundant exception object included in logging.exception call
(TRY401)
607-607: Redundant exception object included in logging.exception call
(TRY401)
618-618: Redundant exception object included in logging.exception call
(TRY401)
630-630: Redundant exception object included in logging.exception call
(TRY401)
662-662: Do not perform function call Depends in argument defaults; instead, perform the call within the function, or read the default from a module-level singleton variable
(B008)
src/core.py
293-293: Avoid specifying long messages outside the exception class
(TRY003)
298-300: Avoid specifying long messages outside the exception class
(TRY003)
391-391: Redundant exception object included in logging.exception call
(TRY401)
394-394: Consider moving this statement to an else block
(TRY300)
401-401: Redundant exception object included in logging.exception call
(TRY401)
405-405: Redundant exception object included in logging.exception call
(TRY401)
408-408: Redundant exception object included in logging.exception call
(TRY401)
🔇 Additional comments (7)
src/main.py (2)
506-510: Acknowledge security scanner warning: SHA256 usage is appropriate here.GitHub Advanced Security flags SHA256 as weak for password hashing. However, this is a false positive in this context. The code correctly uses SHA256 to generate a deterministic identifier from the API key to enforce the one-IP-per-token policy. The API key itself remains the secret; the hash is only used for identification, not for security. This is an appropriate use of SHA256.
Based on past review comments from static analysis.
514-516: LGTM: Token ID integration.The token_id is correctly passed to the core whitelisting function, enabling the one-IP-per-token behavior.
src/core.py (5)
201-203: LGTM: Path normalization addresses security concerns.The addition of
.resolve()to normalize whitelist paths addresses the path injection concerns flagged by CodeQL and previous reviews. This prevents directory traversal attacks by resolving..components and normalizing the path before file operations.Based on past review comments.
Also applies to: 226-228, 303-305, 420-422
133-145: LGTM: Backward-compatible whitelist checking.The updated logic correctly handles both legacy entries (integer expiry) and new token-based entries (dict with expiry and token_id), maintaining backward compatibility while supporting the new device token feature.
234-242: LGTM: Expiry extraction helper for size limiting.The helper function cleanly extracts expiry timestamps from both legacy (int) and new (dict) formats, ensuring the size-limiting logic works correctly across both entry types.
312-336: LGTM: Token replacement logic implements device token behavior.The logic correctly implements one-IP-per-token behavior:
- Searches for existing entries with the same token_id
- Removes the old IP if different from the new one
- Returns the removed IP for firewall rule cleanup
- Maintains backward compatibility by storing token-less entries as integers
The use of
list(whitelist.items())prevents modification during iteration, and breaking after the first match is appropriate since each token_id should have at most one entry.
382-392: LGTM: Firewall rule cleanup for replaced IPs.The logic correctly removes firewall rules for the old IP when a token's whitelisted IP is replaced. The defensive error handling ensures that failures in rule removal don't prevent the overall operation from succeeding.
|
I've added
No manual migration script is required; the system will naturally migrate active users to the new format as they continue to use the service. |
|
integration test updates are still missing |
This PR implements the "Device Tokens" feature, ensuring that each API token can only have one active whitelisted IP address at a time. When a new IP is whitelisted using a token that already has an active entry, the old entry is automatically removed.
Changes
src/core.pyto support storingtoken_idin the whitelist entries.src/core.pyadd_ip_to_whitelistto check for existing entries with the sametoken_idand remove them.src/main.pyto generate a deterministictoken_id(hash of the API key) and pass it to the core logic.tests/test_device_tokens.pyto verify the new behavior and ensure backward compatibility.Sequence Diagram
Before (Legacy Behavior)
sequenceDiagram participant Client participant API participant Whitelist Client->>API: Knock(IP=1.2.3.4, Token=A) API->>Whitelist: Add(1.2.3.4) Whitelist-->>API: OK API-->>Client: 200 OK Client->>API: Knock(IP=5.6.7.8, Token=A) API->>Whitelist: Add(5.6.7.8) Note right of Whitelist: Both 1.2.3.4 and 5.6.7.8 are now active for Token A Whitelist-->>API: OK API-->>Client: 200 OKAfter (Device Tokens)
sequenceDiagram participant Client participant API participant Whitelist participant Firewalld Client->>API: Knock(IP=1.2.3.4, Token=A) API->>API: Generate token_id = Hash(Token=A) API->>Whitelist: Add(IP=1.2.3.4, token_id=Hash(A)) Whitelist-->>API: OK API-->>Client: 200 OK Client->>API: Knock(IP=5.6.7.8, Token=A) API->>API: Generate token_id = Hash(Token=A) API->>Whitelist: Add(IP=5.6.7.8, token_id=Hash(A)) Whitelist->>Whitelist: Find existing entry with token_id=Hash(A) Whitelist->>Whitelist: Remove old IP (1.2.3.4) Whitelist->>Firewalld: Remove Rule(1.2.3.4) Whitelist->>Whitelist: Save new IP (5.6.7.8) Whitelist-->>API: OK (Old IP removed) API-->>Client: 200 OKCloses #21
Summary by CodeRabbit
New Features
Improvements
Tests
Documentation
✏️ Tip: You can customize this high-level summary in your review settings.