-
Notifications
You must be signed in to change notification settings - Fork 0
feat(chatmock): add live prompt controls and reasoning compat #3
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
354be58
9b0afd3
5843866
fe0e209
f5f8f97
0a4ea84
a0edbc0
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,9 +1,14 @@ | ||
| from __future__ import annotations | ||
|
|
||
| from flask import Flask, jsonify | ||
| import hmac | ||
| import ipaddress | ||
| import os | ||
|
|
||
| from flask import Flask, jsonify, request | ||
| from flask_sock import Sock | ||
| from werkzeug.exceptions import BadRequest, UnsupportedMediaType | ||
|
|
||
| from .config import BASE_INSTRUCTIONS, GPT5_CODEX_INSTRUCTIONS | ||
| from .config import get_prompt_manager | ||
| from .http import build_cors_headers | ||
| from .routes_openai import openai_bp | ||
| from .routes_ollama import ollama_bp | ||
|
|
@@ -21,8 +26,16 @@ def create_app( | |
| expose_reasoning_models: bool = False, | ||
| default_web_search: bool = False, | ||
| inject_default_instructions: bool = True, | ||
| prompt_dir: str | None = None, | ||
| prompt_config_path: str | None = None, | ||
| admin_token: str | None = None, | ||
| ) -> Flask: | ||
| app = Flask(__name__) | ||
| prompt_manager = get_prompt_manager( | ||
| prompt_dir=prompt_dir, | ||
| prompt_config_path=prompt_config_path, | ||
| reset=True, | ||
| ) | ||
|
|
||
| app.config.update( | ||
| VERBOSE=bool(verbose), | ||
|
|
@@ -32,18 +45,103 @@ def create_app( | |
| REASONING_COMPAT=reasoning_compat, | ||
| FAST_MODE=bool(fast_mode), | ||
| DEBUG_MODEL=debug_model, | ||
| BASE_INSTRUCTIONS=BASE_INSTRUCTIONS, | ||
| GPT5_CODEX_INSTRUCTIONS=GPT5_CODEX_INSTRUCTIONS, | ||
| EXPOSE_REASONING_MODELS=bool(expose_reasoning_models), | ||
| DEFAULT_WEB_SEARCH=bool(default_web_search), | ||
| INJECT_DEFAULT_INSTRUCTIONS=bool(inject_default_instructions), | ||
| PROMPT_MANAGER=prompt_manager, | ||
| ADMIN_TOKEN=( | ||
| admin_token | ||
| if isinstance(admin_token, str) and admin_token | ||
| else os.getenv("CHATMOCK_ADMIN_TOKEN") or os.getenv("CHATGPT_LOCAL_ADMIN_TOKEN") or None | ||
| ), | ||
| ) | ||
|
|
||
| @app.get("/") | ||
| @app.get("/health") | ||
| def health(): | ||
| return jsonify({"status": "ok"}) | ||
|
|
||
| def _require_local_admin(): | ||
| remote_addr = request.remote_addr | ||
| allow_admin_external = os.getenv("CHATMOCK_ALLOW_ADMIN_EXTERNAL", "false").lower() == "true" | ||
| allowed_local_addresses = {"127.0.0.1", "::1"} | ||
| allowed_bridge_addresses = {"172.17.0.1", "172.18.0.1"} | ||
| trusted_remote_spec = os.getenv("CHATMOCK_ADMIN_TRUSTED_IPS", "").strip() | ||
|
|
||
| def _is_in_trusted_ranges(value: str | None) -> bool: | ||
| if not isinstance(value, str) or not value or not trusted_remote_spec: | ||
| return False | ||
| try: | ||
| remote_ip = ipaddress.ip_address(value) | ||
| except ValueError: | ||
| return False | ||
| for token in (item.strip() for item in trusted_remote_spec.split(",")): | ||
| if not token: | ||
| continue | ||
| try: | ||
| if "/" in token: | ||
| if remote_ip in ipaddress.ip_network(token, strict=False): | ||
| return True | ||
| elif remote_ip == ipaddress.ip_address(token): | ||
| return True | ||
| except ValueError: | ||
| continue | ||
| return False | ||
|
Comment on lines
+64
to
+89
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The allow_admin_external = os.getenv("CHATMOCK_ALLOW_ADMIN_EXTERNAL", "false").lower() == "true"
trusted_networks = []
for token in (t.strip() for t in os.getenv("CHATMOCK_ADMIN_TRUSTED_IPS", "").split(",") if t.strip()):
try:
trusted_networks.append(ipaddress.ip_network(token, strict=False))
except ValueError:
continue
def _require_local_admin():
remote_addr = request.remote_addr
allowed_local_addresses = {"127.0.0.1", "::1"}
allowed_bridge_addresses = {"172.17.0.1", "172.18.0.1"}
def _is_in_trusted_ranges(value: str | None) -> bool:
if not isinstance(value, str) or not value or not trusted_networks:
return False
try:
remote_ip = ipaddress.ip_address(value)
return any(remote_ip in net for net in trusted_networks)
except ValueError:
return False |
||
|
|
||
| is_allowed_bridge = isinstance(remote_addr, str) and remote_addr in allowed_bridge_addresses | ||
| is_local = isinstance(remote_addr, str) and remote_addr in allowed_local_addresses | ||
| is_trusted_remote = _is_in_trusted_ranges(remote_addr) | ||
| if not (is_local or is_allowed_bridge or is_trusted_remote or allow_admin_external): | ||
| return jsonify({"error": {"message": "Admin routes are local-only"}}), 403 | ||
| expected_token = app.config.get("ADMIN_TOKEN") | ||
| provided_token = ( | ||
| request.headers.get("X-ChatMock-Admin-Token") | ||
| or request.headers.get("X-Admin-Token") | ||
| or request.headers.get("Authorization", "").removeprefix("Bearer ").strip() | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The |
||
| ) | ||
| if allow_admin_external and not (is_local or is_allowed_bridge or is_trusted_remote): | ||
| if not (isinstance(expected_token, str) and expected_token): | ||
| return jsonify({"error": {"message": "External admin access requires CHATMOCK_ADMIN_TOKEN"}}), 403 | ||
| if isinstance(expected_token, str) and expected_token: | ||
| if not isinstance(provided_token, str) or not hmac.compare_digest(provided_token, expected_token): | ||
| return jsonify({"error": {"message": "Invalid admin token"}}), 403 | ||
|
Comment on lines
+105
to
+107
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
The new admin routes are writable ( Useful? React with 👍 / 👎. |
||
| return None | ||
|
|
||
| @app.get("/admin/prompts") | ||
| def admin_prompts_state(): | ||
| denied = _require_local_admin() | ||
| if denied is not None: | ||
| return denied | ||
| return jsonify(prompt_manager.as_dict()) | ||
|
|
||
| @app.post("/admin/prompts/reload") | ||
| def admin_prompts_reload(): | ||
| denied = _require_local_admin() | ||
| if denied is not None: | ||
| return denied | ||
| try: | ||
| prompt_manager.reload() | ||
| except (FileNotFoundError, ValueError, OSError, PermissionError) as exc: | ||
| return jsonify({"error": {"message": str(exc)}}), 400 | ||
| return jsonify(prompt_manager.as_dict()) | ||
|
Comment on lines
+117
to
+126
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The |
||
|
|
||
| @app.post("/admin/prompts/config") | ||
| def admin_prompts_config(): | ||
| denied = _require_local_admin() | ||
| if denied is not None: | ||
| return denied | ||
| try: | ||
| payload = request.get_json(silent=False) | ||
| except (BadRequest, UnsupportedMediaType): | ||
| return jsonify({"error": {"message": "Invalid JSON body"}}), 400 | ||
| if not isinstance(payload, dict): | ||
| return jsonify({"error": {"message": "Invalid JSON body"}}), 400 | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| try: | ||
| prompt_manager.update_config(payload) | ||
| except (FileNotFoundError, ValueError, OSError, PermissionError) as exc: | ||
| return jsonify({"error": {"message": str(exc)}}), 400 | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| return jsonify(prompt_manager.as_dict()) | ||
|
|
||
| @app.after_request | ||
| def _cors(resp): | ||
| for k, v in build_cors_headers().items(): | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hardcoding specific Docker bridge gateway addresses like
172.17.0.1and172.18.0.1is fragile, as Docker network ranges can vary significantly depending on the host configuration and the number of existing networks. While theCHATMOCK_ALLOW_ADMIN_EXTERNALescape hatch exists, consider allowing a configurable list of CIDR ranges or trusted gateway addresses via an environment variable to make this more robust for diverse container environments.