339 lines
12 KiB
Python
339 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
VEGA — API-first utility/developer tool host.
|
|
|
|
Every capability is a plain function registered in TOOLS and auto-exposed as:
|
|
GET /api/tools -> catalog {name: {label, category, params, desc}}
|
|
GET /api/<tool>?<params> -> run one tool, return JSON {tool, result}
|
|
POST /api/<tool> -> same, params as JSON body
|
|
GET /health -> liveness + tool count
|
|
|
|
Design goals:
|
|
- API-first: the web UI (later) and the MCP wrapper (later) both consume these
|
|
exact endpoints — no separate logic paths.
|
|
- Deterministic + stdlib-only for the core: no keys, no network, fast, safe.
|
|
- One registry -> N surfaces (REST now, MCP + Hyperion metering next).
|
|
|
|
This is the Phase-1 slice (developer tools). Later phases add internet/infra
|
|
(re-using osint.py's 454), files/media, monitoring, and AI/LLM categories.
|
|
"""
|
|
import base64
|
|
import binascii
|
|
import csv
|
|
import hashlib
|
|
import html
|
|
import io
|
|
import json
|
|
import re
|
|
import time
|
|
import unicodedata
|
|
import uuid as _uuid
|
|
from urllib.parse import parse_qs, quote, unquote, urlparse
|
|
|
|
from flask import Flask, jsonify, request
|
|
|
|
app = Flask(__name__)
|
|
TOOLS = {}
|
|
|
|
|
|
def tool(name, category, params, desc):
|
|
"""Register a function as a tool. params = list of input key names."""
|
|
def deco(fn):
|
|
TOOLS[name] = {"fn": fn, "category": category, "params": params,
|
|
"desc": desc, "label": name.replace("_", " ").title()}
|
|
return fn
|
|
return deco
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# JSON / data
|
|
# ---------------------------------------------------------------------------
|
|
@tool("json_format", "data", ["json"], "Validate + pretty-print JSON.")
|
|
def json_format(json="{}"):
|
|
d = json_loads(json)
|
|
return {"valid": True, "pretty": json.dumps(d, indent=2, sort_keys=True),
|
|
"type": type(d).__name__}
|
|
|
|
|
|
@tool("json_to_csv", "data", ["json"], "Flatten a JSON array of objects to CSV.")
|
|
def json_to_csv(json="[]"):
|
|
d = json_loads(json)
|
|
if not isinstance(d, list):
|
|
return {"error": "expected a JSON array of objects"}
|
|
out = io.StringIO()
|
|
w = csv.DictWriter(out, fieldnames=sorted({k for o in d if isinstance(o, dict) for k in o}))
|
|
w.writeheader()
|
|
for o in d:
|
|
if isinstance(o, dict):
|
|
w.writerow(o)
|
|
return {"csv": out.getvalue()}
|
|
|
|
|
|
@tool("json_diff", "data", ["a", "b"], "Diff two JSON values (added/removed/changed keys).")
|
|
def json_diff(a="{}", b="{}"):
|
|
da, db = json_loads(a), json_loads(b)
|
|
return diff(da, db)
|
|
|
|
|
|
@tool("jwt_decode", "data", ["token"], "Decode a JWT header+payload (no signature verification).")
|
|
def jwt_decode(token=""):
|
|
parts = token.split(".")
|
|
if len(parts) < 2:
|
|
return {"error": "not a JWT (need header.payload.signature)"}
|
|
out = {}
|
|
for name, p in (("header", parts[0]), ("payload", parts[1])):
|
|
try:
|
|
pad = p + "=" * (-len(p) % 4)
|
|
out[name] = json.loads(base64.urlsafe_b64decode(pad).decode("utf-8", "replace"))
|
|
except Exception as e:
|
|
out[name] = {"error": str(e)}
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# encoding / crypto / ids
|
|
# ---------------------------------------------------------------------------
|
|
@tool("hash", "encoding", ["text", "algo"],
|
|
"Hash a string. algo = md5/sha1/sha256/sha512.")
|
|
def hash_text(text="", algo="sha256"):
|
|
h = hashlib.new(algo if algo in hashlib.algorithms_available else "sha256")
|
|
h.update(text.encode())
|
|
return {"algo": h.name, "hex": h.hexdigest(), "bytes": h.digest_size}
|
|
|
|
|
|
@tool("base64_encode", "encoding", ["text"], "Base64-encode text.")
|
|
def b64_encode(text=""):
|
|
return {"encoded": base64.b64encode(text.encode()).decode()}
|
|
|
|
|
|
@tool("base64_decode", "encoding", ["data"], "Base64-decode (returns text + hex).")
|
|
def b64_decode(data=""):
|
|
raw = base64.b64decode(data + "=" * (-len(data) % 4))
|
|
return {"text": raw.decode("utf-8", "replace"), "hex": raw.hex()}
|
|
|
|
|
|
@tool("url_encode", "encoding", ["text"], "Percent-encode a string.")
|
|
def url_encode(text=""):
|
|
return {"encoded": quote(text, safe="")}
|
|
|
|
|
|
@tool("url_decode", "encoding", ["data"], "Percent-decode a string.")
|
|
def url_decode(data=""):
|
|
return {"decoded": unquote(data)}
|
|
|
|
|
|
@tool("url_parse", "encoding", ["url"], "Parse a URL into scheme/host/path/query/fragment.")
|
|
def url_parse(url=""):
|
|
p = urlparse(url)
|
|
return {"scheme": p.scheme, "netloc": p.netloc, "host": p.hostname,
|
|
"port": p.port, "path": p.path, "query": parse_qs(p.query),
|
|
"fragment": p.fragment}
|
|
|
|
|
|
@tool("uuid", "ids", [], "Generate a random UUID v4.")
|
|
def uuid_gen():
|
|
return {"uuid": str(_uuid.uuid4()), "hex": _uuid.uuid4().hex}
|
|
|
|
|
|
@tool("hex_to_text", "encoding", ["hex"], "Decode hex bytes to text.")
|
|
def hex_to_text(hex=""):
|
|
try:
|
|
raw = bytes.fromhex(hex)
|
|
except ValueError as e:
|
|
return {"error": str(e)}
|
|
return {"text": raw.decode("utf-8", "replace"), "bytes": len(raw)}
|
|
|
|
|
|
@tool("text_to_hex", "encoding", ["text"], "Encode text to hex.")
|
|
def text_to_hex(text=""):
|
|
return {"hex": text.encode().hex()}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# text / regex / string utils
|
|
# ---------------------------------------------------------------------------
|
|
@tool("regex_test", "text", ["pattern", "text"], "Test a regex; return matches + groups.")
|
|
def regex_test(pattern="", text=""):
|
|
try:
|
|
rx = re.compile(pattern)
|
|
except re.error as e:
|
|
return {"error": f"bad regex: {e}"}
|
|
matches = [{"match": m.group(0), "groups": list(m.groups()), "span": m.span()}
|
|
for m in rx.finditer(text)]
|
|
return {"valid": True, "count": len(matches), "matches": matches[:50]}
|
|
|
|
|
|
@tool("cron_explain", "text", ["expr"], "Explain a 5-field cron expression.")
|
|
def cron_explain(expr=""):
|
|
fields = expr.split()
|
|
if len(fields) != 5:
|
|
return {"error": "need 5 fields: min hour dom mon dow"}
|
|
names = ["minute", "hour", "day-of-month", "month", "day-of-week"]
|
|
return {names[i]: f for i, f in enumerate(fields)}
|
|
|
|
|
|
@tool("slugify", "text", ["text"], "Slugify text (lowercase, dashes, ascii).")
|
|
def slugify(text=""):
|
|
s = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode()
|
|
s = re.sub(r"[^a-zA-Z0-9]+", "-", s).strip("-").lower()
|
|
return {"slug": s}
|
|
|
|
|
|
@tool("case_convert", "text", ["text", "to"],
|
|
"Convert case: camel/snake/kebab/pascal/title/upper/lower.")
|
|
def case_convert(text="", to="snake"):
|
|
words = [w for w in re.split(r"[^a-zA-Z0-9]+|(?<=[a-z0-9])(?=[A-Z])", text) if w]
|
|
if to == "camel":
|
|
return {"result": words[0].lower() + "".join(w.capitalize() for w in words[1:]) if words else ""}
|
|
if to == "pascal":
|
|
return {"result": "".join(w.capitalize() for w in words)}
|
|
if to == "snake":
|
|
return {"result": "_".join(w.lower() for w in words)}
|
|
if to == "kebab":
|
|
return {"result": "-".join(w.lower() for w in words)}
|
|
if to == "upper":
|
|
return {"result": text.upper()}
|
|
if to == "lower":
|
|
return {"result": text.lower()}
|
|
if to == "title":
|
|
return {"result": text.title()}
|
|
return {"error": f"unknown 'to': {to}"}
|
|
|
|
|
|
@tool("string_metrics", "text", ["text"], "Length/word/line/char counts + Shannon entropy.")
|
|
def string_metrics(text=""):
|
|
if not text:
|
|
return {"chars": 0, "words": 0, "lines": 0, "bytes": 0, "entropy": 0.0}
|
|
import math
|
|
freq = {}
|
|
for c in text:
|
|
freq[c] = freq.get(c, 0) + 1
|
|
n = len(text)
|
|
entropy = -sum((v / n) * math.log2(v / n) for v in freq.values())
|
|
return {"chars": n, "words": len(text.split()), "lines": text.count("\n") + 1,
|
|
"bytes": len(text.encode()), "unique_chars": len(freq),
|
|
"entropy": round(entropy, 3)}
|
|
|
|
|
|
@tool("html_decode", "text", ["html"], "Decode HTML entities.")
|
|
def html_decode(html=""):
|
|
return {"text": html.unescape(html)}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# numbers / time
|
|
# ---------------------------------------------------------------------------
|
|
@tool("base_convert", "numbers", ["value", "from_base", "to_base"],
|
|
"Convert a number between bases 2-36.")
|
|
def base_convert(value="", from_base="10", to_base="16"):
|
|
try:
|
|
n = int(value, int(from_base))
|
|
except ValueError:
|
|
return {"error": "invalid value for base"}
|
|
chars = "0123456789abcdefghijklmnopqrstuvwxyz"
|
|
if not 2 <= int(to_base) <= 36:
|
|
return {"error": "to_base must be 2-36"}
|
|
if n == 0:
|
|
return {"result": "0"}
|
|
neg = n < 0
|
|
n = abs(n)
|
|
out = ""
|
|
while n:
|
|
out = chars[n % int(to_base)] + out
|
|
n //= int(to_base)
|
|
return {"result": ("-" if neg else "") + out,
|
|
"decimal": int(value, int(from_base))}
|
|
|
|
|
|
@tool("epoch", "time", ["ts"], "Convert a unix epoch to human-readable UTC + relative.")
|
|
def epoch(ts=""):
|
|
try:
|
|
t = int(float(ts))
|
|
except ValueError:
|
|
return {"error": "invalid epoch"}
|
|
return {"utc": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(t)),
|
|
"local": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(t)),
|
|
"age_seconds": int(time.time() - t)}
|
|
|
|
|
|
@tool("now", "time", [], "Current unix time (UTC).")
|
|
def now():
|
|
t = time.time()
|
|
return {"epoch": int(t), "utc": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(t))}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# helpers
|
|
# ---------------------------------------------------------------------------
|
|
def json_loads(s):
|
|
try:
|
|
return json.loads(s)
|
|
except Exception as e:
|
|
return {"error": f"invalid JSON: {e}"}
|
|
|
|
|
|
def diff(a, b, path=""):
|
|
out = {"added": [], "removed": [], "changed": []}
|
|
if type(a) is not type(b):
|
|
out["changed"].append(path or "<root>")
|
|
return out
|
|
if isinstance(a, dict):
|
|
for k in set(a) | set(b):
|
|
p = f"{path}.{k}" if path else k
|
|
if k not in b:
|
|
out["removed"].append(p)
|
|
elif k not in a:
|
|
out["added"].append(p)
|
|
elif a[k] != b[k]:
|
|
d = diff(a[k], b[k], p)
|
|
out["added"] += d["added"]; out["removed"] += d["removed"]; out["changed"] += d["changed"]
|
|
elif isinstance(a, list):
|
|
if len(a) != len(b):
|
|
out["changed"].append(f"{path or '<root>'}[length {len(a)}->{len(b)}]")
|
|
for i, (x, y) in enumerate(zip(a, b)):
|
|
if x != y:
|
|
d = diff(x, y, f"{path}[{i}]")
|
|
out["added"] += d["added"]; out["removed"] += d["removed"]; out["changed"] += d["changed"]
|
|
elif a != b:
|
|
out["changed"].append(path or "<root>")
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# routes
|
|
# ---------------------------------------------------------------------------
|
|
@app.route("/health")
|
|
def health():
|
|
return jsonify({"ok": True, "tools": len(TOOLS),
|
|
"categories": sorted({t["category"] for t in TOOLS.values()})})
|
|
|
|
|
|
@app.route("/api/tools")
|
|
def api_catalog():
|
|
return jsonify({name: {"label": t["label"], "category": t["category"],
|
|
"params": t["params"], "desc": t["desc"]}
|
|
for name, t in TOOLS.items()})
|
|
|
|
|
|
@app.route("/api/<name>", methods=["GET", "POST"])
|
|
def api_run(name):
|
|
if name not in TOOLS:
|
|
return jsonify({"error": f"unknown tool: {name}"}), 404
|
|
t = TOOLS[name]
|
|
if request.method == "POST":
|
|
args = request.get_json(silent=True) or {}
|
|
else:
|
|
args = {k: v for k, v in request.args.items()}
|
|
args = {k: args.get(k, "") for k in t["params"]}
|
|
try:
|
|
result = t["fn"](**args)
|
|
except Exception as e:
|
|
return jsonify({"tool": name, "error": f"{type(e).__name__}: {e}"}), 400
|
|
if isinstance(result, dict) and "error" in result:
|
|
return jsonify({"tool": name, **result}), 200
|
|
return jsonify({"tool": name, "result": result})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=8080)
|