Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Security: Enforce mandatory API Keys for V3, fix search logic, and update frontend guide

Browse Source
This commit is contained in:
Scott Idem
2026-01-19 14:11:13 -05:00
parent d8b0c3b0a4
commit cad0d2e867
5 changed files with 325 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
app/lib_sql_search.py
View File

@@ -166,11 +166,23 @@ def sql_search_qry_part(
data[p_name] = query_node.query_string
elif searchable_fields:
like_clauses = []
# Fields to exclude from a generic text 'q' search (numeric, technical, or date fields)
exclude_patterns = [
'enable', 'hide', 'priority', 'sort', 'group',
'created_on', 'updated_on'
]
for field in searchable_fields:
if not any(x in field for x in ['_id', 'enable', 'hide', 'priority', 'sort', 'group', 'created_on', 'updated_on']):
f_p_name = get_param_name()
like_clauses.append(f"`{field}` LIKE :{f_p_name}")
data[f_p_name] = f"%{query_node.query_string}%"
# Exclude exact internal integer IDs (ending in _id)
if field.endswith('_id') or field == 'id':
continue
# Exclude other technical/meta fields
if any(x == field for x in exclude_patterns):
continue
f_p_name = get_param_name()
like_clauses.append(f"`{field}` LIKE :{f_p_name}")
data[f_p_name] = f"%{query_node.query_string}%"
if like_clauses: clauses.append(f"({' OR '.join(like_clauses)})")
for filter_attr in ['and_filters', 'or_filters']:
if hasattr(query_node, filter_attr) and getattr(query_node, filter_attr):

127
app/routers/dependencies_v3.py
View File

@@ -12,35 +12,63 @@ log = logging.getLogger(__name__)
def get_account_context_optional(
x_account_id: Optional[str] = Header(None, min_length=11, max_length=22),
x_no_account_id: Optional[str] = Header(None, min_length=3, max_length=100),
x_no_account_id_token: Optional[str] = Query(None, min_length=11, max_length=22),
x_no_account_id_token: Optional[str] = Query(None, alias='jwt', min_length=11, max_length=22),
x_aether_api_key: Optional[str] = Header(None, min_length=11, max_length=22),
) -> AccountContext:
"""
Resolves the account context but does not raise 403 on failure.
Resolves the account context and enforces API Key validation.
Uses DEFERRED imports to prevent circular dependency at startup.
"""
from app.db_sql import redis_lookup_id_random
from app.db_sql import redis_lookup_id_random, sql_select
from datetime import datetime
resolved_account_id = None
resolved_account_id_random = None
auth_method = 'guest'
api_key_authorized = False
if x_account_id:
resolved_account_id_random = x_account_id
if looked_up_id := redis_lookup_id_random(table_name='account', record_id_random=x_account_id):
resolved_account_id = looked_up_id
auth_method = 'legacy_header'
elif x_no_account_id_token:
resolved_account_id_random = x_no_account_id_token
if looked_up_id := redis_lookup_id_random(table_name='account', record_id_random=x_no_account_id_token):
resolved_account_id = looked_up_id
auth_method = 'token_query'
elif x_no_account_id:
resolved_account_id = None
resolved_account_id_random = '--- NO ACCOUNT ---'
auth_method = 'bypass'
# 1. Mandatory Machine Auth (API Key)
# This identifies the script/app, regardless of the user/account context.
if x_aether_api_key:
sql = "SELECT * FROM api_key WHERE (public_key = :key OR secret_key = :key) LIMIT 1"
if api_key_rec := sql_select(sql=sql, data={'key': x_aether_api_key}):
if api_key_rec.get('enable'):
now = datetime.now()
enable_from = api_key_rec.get('enable_from')
enable_to = api_key_rec.get('enable_to')
if (not enable_from or enable_from <= now) and (not enable_to or now <= enable_to):
api_key_authorized = True
else:
log.warning(f"Security: API Key {x_aether_api_key} expired/not yet valid.")
else:
log.warning(f"Security: API Key {x_aether_api_key} is disabled.")
else:
log.warning(f"Security: API Key {x_aether_api_key} not found.")
# 2. Context Resolution (Only if API Key is valid)
if api_key_authorized:
# A. Resolve via Account ID Header
if x_account_id:
resolved_account_id_random = x_account_id
if looked_up_id := redis_lookup_id_random(table_name='account', record_id_random=x_account_id):
resolved_account_id = looked_up_id
auth_method = 'legacy_header'
# B. Resolve via JWT / Token Query Param
elif x_no_account_id_token:
resolved_account_id_random = x_no_account_id_token
if looked_up_id := redis_lookup_id_random(table_name='account', record_id_random=x_no_account_id_token):
resolved_account_id = looked_up_id
auth_method = 'token_query'
# C. Resolve via Administrative Bypass
elif x_no_account_id and x_no_account_id.lower() not in ['false', '0', 'null', 'undefined', 'none', 'no_account_id_here']:
resolved_account_id = None
resolved_account_id_random = '--- NO ACCOUNT ---'
auth_method = 'bypass'
return AccountContext(
account_id=resolved_account_id,
account_id=resolved_account_id,
account_id_random=resolved_account_id_random,
auth_method=auth_method,
administrator=(auth_method == 'bypass'),
@@ -51,10 +79,67 @@ def get_account_context_optional(
def get_account_context(
x_account_id: Optional[str] = Header(None, min_length=11, max_length=22),
x_no_account_id: Optional[str] = Header(None, min_length=3, max_length=100),
x_no_account_id_token: Optional[str] = Query(None, min_length=11, max_length=22),
x_no_account_id_token: Optional[str] = Query(None, alias='jwt', min_length=11, max_length=22),
x_aether_api_key: Optional[str] = Header(None, min_length=11, max_length=22),
) -> AccountContext:
"""Strict version of account context resolution."""
ctx = get_account_context_optional(x_account_id, x_no_account_id, x_no_account_id_token)
ctx = get_account_context_optional(x_account_id, x_no_account_id, x_no_account_id_token, x_aether_api_key)
if ctx.auth_method == 'guest':
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='Account context required.')
return ctx
# --- Shared Pagination & Status Dependencies ---
class PaginationParams:
def __init__(
self,
limit: int = Query(100, ge=0),
offset: int = Query(0, ge=0),
):
self.limit = limit
self.offset = offset
class StatusFilterParams:
def __init__(
self,
enabled: str = Query('enabled'),
hidden: str = Query('not_hidden'),
):
self.enabled = enabled
self.hidden = hidden
class SerializationParams:
def __init__(
self,
by_alias: bool = Query(True),
exclude_unset: bool = Query(False),
exclude_defaults: bool = Query(False),
exclude_none: bool = Query(False),
):
self.by_alias = by_alias
self.exclude_unset = exclude_unset
self.exclude_defaults = exclude_defaults
self.exclude_none = exclude_none
class DelayParams:
def __init__(
self,
x_delay_ms: Optional[int] = Header(0, alias='X-Delay-ms'),
delay_ms: Optional[int] = Query(0),
):
val = max(x_delay_ms or 0, delay_ms or 0)
self.sleep_time_ms = val
self.sleep_time_s = val / 1000.0
def get_account_context(
x_account_id: Optional[str] = Header(None, min_length=11, max_length=22),
x_no_account_id: Optional[str] = Header(None, min_length=3, max_length=100),
x_no_account_id_token: Optional[str] = Query(None, alias='jwt', min_length=11, max_length=22),
x_aether_api_key: Optional[str] = Header(None, min_length=11, max_length=22),
) -> AccountContext:
"""Strict version of account context resolution."""
ctx = get_account_context_optional(x_account_id, x_no_account_id, x_no_account_id_token, x_aether_api_key)
if ctx.auth_method == 'guest':
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='Account context required.')
return ctx

59
documentation/PROJECT_SECURITY_HARDENING.md Normal file
View File

@@ -0,0 +1,59 @@
# Project: API Security Hardening (V3)
**Status:** Draft / Planning
**Date:** Jan 18, 2026
**Owner:** Scott / Aether API Team
## 1. Executive Summary
This project aims to close a critical security vulnerability in the Aether API V3 dependencies where the `x_no_account_id` header allows unauthorized "Superuser/Bypass" access without validation. Additionally, it addresses the lack of cryptographic verification for JWTs in the V3 CRUD layer.
## 2. The Vulnerabilities
### A. The "Bypass Header" (Critical)
* **Issue:** The `get_account_context_optional` dependency in `app/routers/dependencies_v3.py` accepts a header `x_no_account_id`. If present, it sets `auth_method = 'bypass'` and grants full administrative privileges (`super=True`).
* **Note:** This header may be sent as `x_no_account_id` or `x-no-account-id`. FastAPI's `Header` logic should handle the conversion, but validation must be explicit.
* **Current State:** There is **no** validation of this header. Sending `x_no_account_id: anything` works.
* **Risk:** Total system compromise via simple header injection.
### B. JWT Verification (High)
* **Issue:** The system validates users by looking up the `x_account_id` string in the database (Redis/MariaDB).
* **Current State:** It does **not** cryptographically verify the JWT signature or expiration in the V3 dependency chain.
* **Risk:** Account impersonation if an Account ID is leaked.
## 3. Implementation Plan
### Phase 1: Secure the Bypass Header (Immediate)
**Goal:** Restrict "Bypass Mode" to clients possessing a valid, active API Key.
1. **Refactor Dependencies:**
* Update `get_account_context_optional` to accept `x_aether_api_key`.
* **Logic Change:** If `x_no_account_id` header is detected:
* **Require** `x_aether_api_key`.
* **Lookup** key in `api_key` table.
* **Verify** `enable = 1` and `now()` is between `enable_from` and `enable_to`.
* **Failure Behavior:** If key is invalid/missing, log warning and deny 'bypass' status (fall back to 'guest').
2. **Verification:**
* Test `curl` with header only -> Expect 403/Forbidden.
* Test `curl` with header + valid Key -> Expect 200/OK.
* Test Frontend File Download (uses `x_no_account_id_token` param) -> Expect 200/OK (No regression).
### Phase 2: JWT Verification (Subsequent)
**Goal:** Replace DB-lookup auth with cryptographic JWT validation.
1. **Audit:** Confirm `app/lib_jwt.py` logic is sound.
2. **Middleware/Dependency:** Integrate `decode_jwt` into the `get_account_context` flow.
3. **Transition:** Allow dual-mode (DB lookup OR JWT) for a transition period if necessary, or cut over if frontend sends valid JWTs.
## 4. Impact Analysis
* **Frontend (SvelteKit):**
* Audit confirmed no usage of `x_no_account_id` **header** in active source code.
* Usage of `x_no_account_id_token` (Query Param) is **safe** and distinct from the header logic.
* **Scripts/External Tools:**
* Any external scripts using the "Bypass" header must be updated to send a valid API Key.
## 5. Action Items
- [ ] Create `PROJECT_SECURITY_HARDENING.md` (This document).
- [ ] Refactor `app/routers/dependencies_v3.py`.
- [ ] Verify fix with `curl` tests.
- [ ] Commit changes.

110
documentation/V3_FRONTEND_API_GUIDE.md
View File

@@ -19,30 +19,38 @@ This guide explains how to update or create frontend functions to interact with
## 2. Authentication and Security (Mandatory in V3)
As of January 2026, the V3 architecture enforces strict **Multi-Tenant Isolation**. Most requests now require valid authentication to prevent data leakage between accounts.
As of January 2026, the V3 architecture enforces strict **Multi-Tenant Isolation** and **Machine Authorization**. Most requests now require two levels of validation.
### A. Authentication Requirement
Almost all V3 CRUD endpoints require a standard Bearer token in the `Authorization` header.
### A. The "Entry Ticket" (API Key)
**Mandatory for all requests.** To prevent unauthorized clients from using the API, every request must provide a valid `x-aether-api-key` in the header. This identifies the application or script (e.g., the Svelte frontend, the Flask legacy app, or an AI agent).
* **Mandatory:** You must provide a valid JWT for nearly all requests.
* **Account Isolation:** The backend automatically filters all results based on the `account_id` found in your JWT. You cannot access data belonging to another account even if you know the random ID.
* **Status Codes:**
* `401 Unauthorized`: Your JWT is invalid or expired.
* `403 Forbidden`: No authentication provided, or you attempted to access an object belonging to a different account.
* **Header:** `x-aether-api-key: <your_app_key>`
* **Status Code:** `403 Forbidden` if missing, invalid, or expired.
**Example Request Header:**
```http
Authorization: Bearer <your_jwt_token>
```
### B. The "Visa" (Account Context)
Once the API Key is validated, you must specify the context of your request.
### B. The "Bootstrap Paradox" Exception (`site_domain`)
1. **Standard User Access**: Provide the `x-account-id` (the random string ID). This restricts all results to that specific tenant.
* **Header:** `x-account-id: <account_id_random>`
2. **Administrative Bypass**: For authorized scripts needing global access, use the bypass header.
* **Header:** `x-no-account-id: bypass`
3. **Authentication Requirement**: Most V3 CRUD endpoints also require a standard Bearer token in the `Authorization` header for user-level actions.
### C. The "Bootstrap Paradox" Exception (`site_domain`)
There is one critical exception to strict authentication: **`site_domain` search**.
Because the frontend needs to lookup the site configuration (to know which account it's on) *before* a user can log in, the following endpoint allows unauthenticated (guest) access:
Because the frontend needs to lookup the site configuration (to know which account it's on) *before* it has an Account ID or User token, the following endpoint allows unauthenticated (**guest**) access:
**Endpoint:** `POST /v3/crud/site_domain/search`
This is the only V3 search allowed without a JWT. All other object types (journal, account, post, etc.) will return `403 Forbidden` if accessed without a token.
This is the only V3 search allowed without an Account ID or JWT. However, it **still requires a valid API Key**.
### D. Fail-Fast on Auth Failures (401/403)
To prevent unnecessary server load and looping, the frontend implements a **Fail-Fast** policy for authentication and authorization errors.
* **401 Unauthorized**: Means the token is missing, invalid, or expired.
* **403 Forbidden**: Means the token is valid, but you do not have permission to access the specific resource (or you attempted a cross-tenant request without bypass).
* **Protocol**: If the API returns a 401 or 403, the frontend **must not retry** the request automatically. It should stop immediately and, if applicable, redirect the user to the login page or display a "Permission Denied" message.
---
@@ -185,6 +193,72 @@ const downloadUrl = `${BASE_URL}/v3/crud/hosted_file/${fileId}/?jwt=${jwtToken}`
## 7. Best Practices for V3
1. **Use `view` for Rich Data**: Instead of manually joining data in separate calls, use `?view=enriched` or `?view=detail`.
2. **Singular Nouns**: Always use singular names for `obj_type` (e.g., `journal`).
3. **Strict Typing**: Ensure your `data` objects match the backend models to avoid `400 Bad Request` validation errors.
---
## 8. Standard Patterns & Common Pitfalls (2026 Update)
To ensure stability across the Aether mesh, all frontend components must adhere to these established patterns.
### A. The "Whitelist" Standard for Saves
**Problem:** Sending the entire object returned by a GET request back to a PATCH endpoint will cause a `400 Bad Request`. This happens because the object contains technical metadata (like `journal_id`, `created_on`, `for_type`) that the API cannot "SET".
**Standard:** When preparing a save payload, explicitly whitelist ONLY the user-editable fields.
```ts
// ✅ CORRECT: Whitelist only editable fields
const data_kv = {
name: tmp_obj.name,
content: tmp_obj.content,
tags: tmp_obj.tags,
category_code: tmp_obj.category_code
};
// ❌ WRONG: Passing the whole object or just blacklisting a few
const data_kv = { ...tmp_obj };
delete data_kv.id; // Still contains other computed columns!
```
### B. The Triple-ID Save Pattern
**Standard:** When sending an ID in a POST/PATCH payload (e.g. creating a child object), use the random string version with the `_id_random` suffix.
* **Property:** `[obj_type]_id_random`
* **Value:** A string (e.g., `qpgvOh5nOYI`)
**Warning:** Sending a string value under an integer key (e.g. `journal_id: "qpgvOh5nOYI"`) will trigger a Pydantic validation error on the backend.
### C. Handle 'NULL' vs '0' for Booleans
**Pitfall:** Fields like `hide` or `priority` can be `NULL` in the database.
**Standard:** Ensure your synchronization logic in components handles `null`, `undefined`, and `0` identically for boolean checks to prevent "ghost" changes being detected by Svelte 5.

52
tests/repro_security_bypass.py Normal file
View File

@@ -0,0 +1,52 @@
import requests
import sys
BASE_URL = "http://dev-api.oneskyit.com" # Standard Dev Domain
OBJ_TYPE = "journal"
ENDPOINT = f"{BASE_URL}/v3/crud/{OBJ_TYPE}/schema"
# Replace with the key found from DB query
VALID_API_KEY = "dummy_key_placeholder"
def test_request(description, headers, expected_status):
print(f"Testing: {description}")
try:
response = requests.get(ENDPOINT, headers=headers, timeout=10)
status = response.status_code
result = "PASS" if status == expected_status else "FAIL"
print(f" Status: {status} (Expected: {expected_status}) -> {result}")
if result == "FAIL":
print(f" Response: {response.text[:200]}...")
return result == "PASS"
except Exception as e:
print(f" Error: {e}")
return False
def main():
if len(sys.argv) > 1:
global VALID_API_KEY
VALID_API_KEY = sys.argv[1]
print(f"--- Security Bypass Test (Target: {ENDPOINT}) ---")
# Case 1: No Auth
# Expected: 403 Forbidden (Account context required)
test_request("No Auth Headers", {}, 403)
# Case 2: Vulnerability Check (Bypass Header Only)
# AFTER FIX: Expected 403 (Protected)
test_request("Bypass Header Only (Vulnerability Check)", {"x-no-account-id": "bypass"}, 403)
# Case 3: Invalid API Key + Bypass Header
# Expected: 403 Forbidden
test_request("Bypass Header + Invalid Key", {"x-no-account-id": "bypass", "x-aether-api-key": "invalid-key-12345"}, 403)
# Case 4: Valid API Key + Bypass Header
# Expected: 200 OK
if VALID_API_KEY != "dummy_key_placeholder":
test_request("Bypass Header + Valid Key", {"x-no-account-id": "bypass", "x-aether-api-key": VALID_API_KEY}, 200)
else:
print("Skipping Case 4 (No Valid API Key provided)")
if __name__ == "__main__":
main()