135 lines
5.2 KiB
Python
135 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
|
|
TOKEN_RE = re.compile(r"{{\s*([a-zA-Z0-9_.-]+)\s*}}")
|
|
|
|
|
|
class HTTPRequestsPlugin:
|
|
name = "HTTP Requests"
|
|
desc = "Send configurable HTTP requests to webhooks, local services, and automation tools."
|
|
version = "0.1.0"
|
|
actions = [
|
|
{
|
|
"id": "send_request",
|
|
"name": "Send Request",
|
|
"desc": "Send a GET, POST, PUT, PATCH, or DELETE request.",
|
|
"fields": [
|
|
{
|
|
"id": "method",
|
|
"label": "Method",
|
|
"type": "select",
|
|
"default": "POST",
|
|
"options": [
|
|
{"label": "GET", "value": "GET"},
|
|
{"label": "POST", "value": "POST"},
|
|
{"label": "PUT", "value": "PUT"},
|
|
{"label": "PATCH", "value": "PATCH"},
|
|
{"label": "DELETE", "value": "DELETE"},
|
|
],
|
|
},
|
|
{
|
|
"id": "url",
|
|
"label": "URL",
|
|
"type": "url",
|
|
"required": True,
|
|
"placeholder": "http://127.0.0.1:8080/webhook",
|
|
},
|
|
{
|
|
"id": "headers",
|
|
"label": "Headers",
|
|
"type": "key_value",
|
|
"default": [{"key": "Content-Type", "value": "application/json"}],
|
|
},
|
|
{
|
|
"id": "body",
|
|
"label": "Body",
|
|
"type": "json",
|
|
"default": '{\n "button": "{{event.button}}",\n "event": "{{event.event}}"\n}',
|
|
},
|
|
{"id": "timeout", "label": "Timeout Seconds", "type": "number", "default": 5},
|
|
{"id": "log_response", "label": "Log response", "type": "boolean", "default": True},
|
|
],
|
|
}
|
|
]
|
|
|
|
def on_load(self, ctx):
|
|
ctx.db.add_event("plugin.loaded", {"plugin": self.name})
|
|
|
|
def execute_action(self, ctx, action_id, config, event):
|
|
if action_id != "send_request":
|
|
raise ValueError(f"Unknown HTTP action '{action_id}'.")
|
|
|
|
method = str(config.get("method", "POST")).upper()
|
|
if method not in {"GET", "POST", "PUT", "PATCH", "DELETE"}:
|
|
raise ValueError(f"Unsupported HTTP method '{method}'.")
|
|
|
|
context = {"event": event or {}}
|
|
url = self._render(str(config.get("url", "")).strip(), context)
|
|
if not url.startswith(("http://", "https://")):
|
|
raise ValueError("HTTP Request URL must start with http:// or https://.")
|
|
|
|
timeout = max(1.0, min(30.0, float(config.get("timeout", 5) or 5)))
|
|
headers = self._headers(config.get("headers"), context)
|
|
body_text = self._render(str(config.get("body", "") or ""), context).strip()
|
|
data = None
|
|
if method not in {"GET", "DELETE"} and body_text:
|
|
data = body_text.encode("utf-8")
|
|
headers.setdefault("Content-Type", "application/json")
|
|
|
|
request = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
response_body = response.read(4096).decode("utf-8", errors="replace")
|
|
status = response.status
|
|
except urllib.error.HTTPError as exc:
|
|
response_body = exc.read(4096).decode("utf-8", errors="replace")
|
|
raise RuntimeError(f"HTTP request failed with {exc.code}: {response_body[:300]}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise RuntimeError(f"HTTP request failed: {exc.reason}") from exc
|
|
|
|
if config.get("log_response", True):
|
|
ctx.db.add_event(
|
|
"plugin.http_response",
|
|
{"status": status, "url": url, "method": method, "body": response_body[:1000]},
|
|
)
|
|
|
|
def _headers(self, raw_headers: Any, context: dict[str, Any]) -> dict[str, str]:
|
|
if isinstance(raw_headers, dict):
|
|
items = raw_headers.items()
|
|
elif isinstance(raw_headers, list):
|
|
items = ((item.get("key"), item.get("value")) for item in raw_headers if isinstance(item, dict))
|
|
else:
|
|
items = []
|
|
headers = {}
|
|
for key, value in items:
|
|
rendered_key = self._render(str(key or "").strip(), context)
|
|
if not rendered_key:
|
|
continue
|
|
headers[rendered_key] = self._render(str(value or ""), context)
|
|
return headers
|
|
|
|
def _render(self, template: str, context: dict[str, Any]) -> str:
|
|
def replace(match):
|
|
value: Any = context
|
|
for part in match.group(1).split("."):
|
|
if isinstance(value, dict):
|
|
value = value.get(part, "")
|
|
else:
|
|
value = getattr(value, part, "")
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, (dict, list)):
|
|
return json.dumps(value)
|
|
return str(value)
|
|
|
|
return TOKEN_RE.sub(replace, template)
|
|
|
|
|
|
PLUGIN = HTTPRequestsPlugin()
|