hadoxmin/app.polisplexity.tech
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Zero hardcode in router
Some checks failed
continuous-integration/drone/push Build is failing
Details

Browse Source
This commit is contained in:
Ekaropolus 2025-09-17 01:19:04 -06:00
parent c748a69bd4
commit ea4eac3e30
1 changed files with 3 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
pxy_bots/router.py
View File

@ -6,22 +6,6 @@ from urllib.parse import urlparse
logger = logging.getLogger(__name__)
# --- allowlist of outbound hosts for in-memory routes (kept) ---
ALLOWED_FORWARD_HOSTS = {"127.0.0.1", "localhost", "app.polisplexity.tech"}
# --- in-memory ROUTE_MAP (kept) ---
ROUTE_MAP: Dict[str, Dict[str, str]] = {
"PolisplexityBot": {
"_default": "http://127.0.0.1:8000/api/bots/echo_render",
"_callback": "http://127.0.0.1:8000/api/bots/echo_render",
"report_trash": "http://127.0.0.1:8000/api/bots/echo_render",
},
"*": {
"_default": "http://127.0.0.1:8000/api/bots/echo_render",
"_callback": "http://127.0.0.1:8000/api/bots/echo_render",
},
}
# Try to use requests; fallback to urllib
try:
import requests # type: ignore
@ -31,24 +15,6 @@ except Exception:
_HAS_REQUESTS = False
# -----------------------------
# Helpers (kept + new)
# -----------------------------
def _allowed(url: str) -> Tuple[bool, Optional[str]]:
"""Allowlist check for in-memory routes (ROUTE_MAP)."""
try:
p = urlparse(url)
host = (p.hostname or "").lower()
if p.scheme not in {"http", "https"}:
return False, "bad_scheme"
if host not in ALLOWED_FORWARD_HOSTS:
return False, f"host_not_allowed:{host}"
return True, None
except Exception as e:
return False, f"invalid_url:{e}"
def _is_allowed(url: str, allowed_hosts: Optional[set]) -> Tuple[bool, str]:
"""Allowlist check for DB routes (per-Connection)."""
try:
@ -68,28 +34,8 @@ def _compose_url(base: str, path: str) -> str:
# -----------------------------
# In-memory routing (kept)
# DB routing (Admin-driven)
# -----------------------------
def pick_url(bot_name: str, canon: Dict) -> Optional[str]:
"""Decide target URL from bot + command/trigger using ROUTE_MAP."""
bot_routes = ROUTE_MAP.get(bot_name) or ROUTE_MAP.get("*") or {}
trigger = ((canon.get("command") or {}).get("trigger")) or "message"
cmd = ((canon.get("command") or {}).get("name")) or ""
if trigger == "callback":
return bot_routes.get("_callback") or bot_routes.get("_default")
if cmd:
return bot_routes.get(cmd) or bot_routes.get("_default")
return bot_routes.get("_default")
# -----------------------------
# NEW: DB routing (Admin)
# -----------------------------
def pick_db_route(bot_name: str, canon: Dict) -> Optional[Dict]:
"""
Look up CommandRoute for this bot + trigger/(optional) command.
@ -140,14 +86,13 @@ def pick_db_route(bot_name: str, canon: Dict) -> Optional[Dict]:
# -----------------------------
# HTTP POST (extended)
# HTTP POST (DB routes)
# -----------------------------
def post_json(url: str, payload: Dict, timeout: float = 4.0, headers: Optional[Dict] = None) -> Tuple[int, Dict]:
"""
Blocking POST JSON; never raises.
Returns (status_code, body_json_or_wrapper).
`headers` is optional for DB routes; in-memory callers continue to work.
`headers` is optional for DB routes.
"""
hdrs = {"Content-Type": "application/json", **(headers or {})}
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")