551 lines
21 KiB
Python
551 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""sicabe-config-agent — akses konfigurasi platform SiCABE per host.
|
|
|
|
Control-panel TIDAK mengakses filesystem platform langsung; agent ini yang
|
|
membacanya di host tempat sicabe-doc berada, lalu diekspos lewat HTTP ber-api-key.
|
|
|
|
Endpoint (butuh header `X-Api-Key` kecuali `/healthz`):
|
|
GET /healthz
|
|
GET /env?key=NAME[&key=NAME2] # read-only, whitelist (minimal 1 key)
|
|
GET /catalog # daftar katalog Trino (nilai sensitif di-redact)
|
|
GET /catalog/<name> # detail satu katalog (redact)
|
|
PUT /catalog/<name> # body {"properties": {...}} — tulis (upsert)
|
|
DELETE /catalog/<name> # hapus
|
|
GET /dbt # project dbt terparse (models/sources/DAG)
|
|
GET /ca # PEM sertifikat TLS host (opsional)
|
|
GET /opa/data # dokumen data OPA (config/opa/data.json) — mentah
|
|
PUT /opa/data # body {"content": "<json>"} — validasi + tulis atomic (.bak)
|
|
POST /services/scan # body {"publicHost","types":[...]} — scan service (panel-driven)
|
|
|
|
Env:
|
|
AGENT_API_KEY (wajib) api key yang harus dikirim panel
|
|
AGENT_PORT (default 8080)
|
|
SICABE_ENV_FILE (default /config/.env)
|
|
TRINO_CATALOG_DIR (default /config/config/trino/catalog)
|
|
DBT_PROJECT_DIR (default /config/config/airflow/dbt/sicabe)
|
|
AGENT_CA_FILE (opsional) path PEM untuk GET /ca
|
|
OPA_DATA_FILE (default /config/config/opa/data.json) dokumen data OPA
|
|
DOCKER_PROXY_URL (default http://docker-proxy:2375) sumber daftar container
|
|
"""
|
|
import hmac
|
|
import json
|
|
import os
|
|
import re
|
|
import socket
|
|
import urllib.request
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import urlparse, parse_qs
|
|
|
|
import yaml
|
|
|
|
AGENT_API_KEY = os.environ.get("AGENT_API_KEY", "")
|
|
AGENT_PORT = int(os.environ.get("AGENT_PORT", "8080"))
|
|
ENV_FILE = os.environ.get("SICABE_ENV_FILE", "/config/.env")
|
|
CATALOG_DIR = os.environ.get("TRINO_CATALOG_DIR", "/config/config/trino/catalog")
|
|
DBT_DIR = os.environ.get("DBT_PROJECT_DIR", "/config/config/airflow/dbt/sicabe")
|
|
CA_FILE = os.environ.get("AGENT_CA_FILE", "")
|
|
OPA_DATA_FILE = os.environ.get("OPA_DATA_FILE", "/config/config/opa/data.json")
|
|
# docker-proxy internal (satu stack dengan agent) — untuk scan service tanpa
|
|
# perlu mount docker.sock ke agent.
|
|
DOCKER_PROXY_URL = os.environ.get("DOCKER_PROXY_URL", "http://docker-proxy:2375")
|
|
|
|
VALID_CATALOG = re.compile(r"^[a-z][a-z0-9_-]{0,62}$")
|
|
SENSITIVE = re.compile(r"(password|secret|credential|access-key|apikey|api-key|token)", re.I)
|
|
REF_RE = re.compile(r"""ref\(\s*['"]([^'"]+)['"]\s*\)""")
|
|
SOURCE_RE = re.compile(r"""source\(\s*['"]([^'"]+)['"]\s*,\s*['"]([^'"]+)['"]\s*\)""")
|
|
|
|
|
|
# ── helpers: .env ────────────────────────────────────────────────────────────
|
|
def read_env(keys):
|
|
"""Baca key tertentu dari ENV_FILE (read-only, whitelist)."""
|
|
wanted = set(keys)
|
|
out = {}
|
|
try:
|
|
with open(ENV_FILE, "r", encoding="utf-8") as fh:
|
|
for line in fh:
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
k, v = line.split("=", 1)
|
|
k = k.strip()
|
|
if k not in wanted:
|
|
continue
|
|
v = v.strip()
|
|
if len(v) >= 2 and v[0] == v[-1] and v[0] in "\"'":
|
|
v = v[1:-1]
|
|
out[k] = v
|
|
except OSError:
|
|
return {}
|
|
return out
|
|
|
|
|
|
# ── helpers: katalog Trino ───────────────────────────────────────────────────
|
|
def redact(props):
|
|
return {k: ("••••••••" if SENSITIVE.search(k) else v) for k, v in props.items()}
|
|
|
|
|
|
def parse_properties(content):
|
|
props = {}
|
|
for line in content.splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
k, v = line.split("=", 1)
|
|
props[k.strip()] = v.strip()
|
|
return props
|
|
|
|
|
|
def serialize_properties(props):
|
|
return "".join(f"{k}={v}\n" for k, v in props.items())
|
|
|
|
|
|
def catalog_path(name):
|
|
return os.path.join(CATALOG_DIR, f"{name}.properties")
|
|
|
|
|
|
def list_catalogs():
|
|
out = []
|
|
try:
|
|
names = sorted(os.listdir(CATALOG_DIR))
|
|
except OSError:
|
|
return out
|
|
for fname in names:
|
|
if not fname.endswith(".properties"):
|
|
continue
|
|
name = fname[: -len(".properties")]
|
|
try:
|
|
with open(os.path.join(CATALOG_DIR, fname), "r", encoding="utf-8") as fh:
|
|
props = parse_properties(fh.read())
|
|
except OSError:
|
|
continue
|
|
out.append({"name": name, "connector": props.get("connector.name", "unknown"), "properties": redact(props)})
|
|
return sorted(out, key=lambda c: c["name"])
|
|
|
|
|
|
def get_catalog(name):
|
|
for c in list_catalogs():
|
|
if c["name"] == name:
|
|
return c
|
|
return None
|
|
|
|
|
|
def put_catalog(name, props):
|
|
os.makedirs(CATALOG_DIR, exist_ok=True)
|
|
with open(catalog_path(name), "w", encoding="utf-8") as fh:
|
|
fh.write(serialize_properties(props))
|
|
|
|
|
|
def write_opa_data(content):
|
|
"""Tulis dokumen data OPA secara atomic + simpan isi sebelumnya ke .bak.
|
|
|
|
Atomic (temp + os.replace) agar crash di tengah tulis tidak meninggalkan file
|
|
separuh jadi — OPA GAGAL START bila JSON-nya tidak valid, dan itu mematikan
|
|
authz Trino. Backup `.bak` dipakai untuk rollback (Tahap 4).
|
|
"""
|
|
os.makedirs(os.path.dirname(OPA_DATA_FILE) or ".", exist_ok=True)
|
|
if os.path.exists(OPA_DATA_FILE):
|
|
with open(OPA_DATA_FILE, "rb") as fh:
|
|
previous = fh.read()
|
|
with open(OPA_DATA_FILE + ".bak", "wb") as fh:
|
|
fh.write(previous)
|
|
tmp = OPA_DATA_FILE + ".tmp"
|
|
with open(tmp, "w", encoding="utf-8") as fh:
|
|
fh.write(content)
|
|
fh.flush()
|
|
os.fsync(fh.fileno())
|
|
os.replace(tmp, OPA_DATA_FILE)
|
|
|
|
|
|
def delete_catalog(name):
|
|
path = catalog_path(name)
|
|
if not os.path.exists(path):
|
|
raise FileNotFoundError(name)
|
|
os.unlink(path)
|
|
|
|
|
|
# ── helpers: project dbt (bentuk = lib/dbt.ts) ───────────────────────────────
|
|
def layer_from_path(rel):
|
|
if "bronze_to_silver" in rel:
|
|
return "silver"
|
|
if "silver_to_gold" in rel:
|
|
return "gold"
|
|
if "incremental" in rel:
|
|
return "bronze"
|
|
return "silver"
|
|
|
|
|
|
def load_dbt():
|
|
name, version, profile = "sicabe", "1.0.0", "trino"
|
|
try:
|
|
with open(os.path.join(DBT_DIR, "dbt_project.yml"), "r", encoding="utf-8") as fh:
|
|
yml = fh.read()
|
|
m = re.search(r"^name:\s*['\"]?([^'\"\n]+)", yml, re.M)
|
|
if m:
|
|
name = m.group(1).strip()
|
|
m = re.search(r"^version:\s*['\"]?([^'\"\n]+)", yml, re.M)
|
|
if m:
|
|
version = m.group(1).strip()
|
|
m = re.search(r"^profile:\s*['\"]?([^'\"\n]+)", yml, re.M)
|
|
if m:
|
|
profile = m.group(1).strip()
|
|
except OSError:
|
|
pass
|
|
|
|
models = []
|
|
models_dir = os.path.join(DBT_DIR, "models")
|
|
for root, _dirs, files in os.walk(models_dir):
|
|
for fname in files:
|
|
if not fname.endswith(".sql"):
|
|
continue
|
|
full = os.path.join(root, fname)
|
|
rel = os.path.relpath(full, models_dir)
|
|
try:
|
|
with open(full, "r", encoding="utf-8") as fh:
|
|
sql = fh.read()
|
|
except OSError:
|
|
continue
|
|
layer = layer_from_path(rel)
|
|
models.append({
|
|
"name": fname[: -len(".sql")],
|
|
"path": rel,
|
|
"layer": layer,
|
|
"materialization": "incremental" if "incremental" in rel else "table",
|
|
"schema": layer,
|
|
"dependsOn": REF_RE.findall(sql),
|
|
"sources": [f"{a}.{b}" for a, b in SOURCE_RE.findall(sql)],
|
|
"sql": sql,
|
|
})
|
|
|
|
sources = []
|
|
sources_file = os.path.join(models_dir, "sources.yml")
|
|
if os.path.exists(sources_file):
|
|
try:
|
|
with open(sources_file, "r", encoding="utf-8") as fh:
|
|
doc = yaml.safe_load(fh.read()) or {}
|
|
for s in doc.get("sources", []) or []:
|
|
nm = s.get("name") or ""
|
|
if not nm:
|
|
continue
|
|
sources.append({
|
|
"name": nm,
|
|
"database": s.get("database") or "",
|
|
"schema": s.get("schema") or "",
|
|
"tables": [{"name": t.get("name") or "", "description": t.get("description")}
|
|
for t in (s.get("tables") or [])],
|
|
})
|
|
except (OSError, yaml.YAMLError):
|
|
sources = []
|
|
|
|
nodes, edges = [], []
|
|
for src in sources:
|
|
for tbl in src["tables"]:
|
|
nodes.append({"id": f"source.{src['name']}.{tbl['name']}", "label": f"{src['name']}.{tbl['name']}",
|
|
"type": "source", "layer": "bronze"})
|
|
for m in models:
|
|
nodes.append({"id": f"model.{m['name']}", "label": m["name"], "type": "model", "layer": m["layer"]})
|
|
for dep in m["dependsOn"]:
|
|
edges.append({"id": f"model.{dep}->model.{m['name']}", "source": f"model.{dep}", "target": f"model.{m['name']}"})
|
|
for s in m["sources"]:
|
|
db, _, tbl = s.partition(".")
|
|
edges.append({"id": f"source.{db}.{tbl}->model.{m['name']}", "source": f"source.{db}.{tbl}", "target": f"model.{m['name']}"})
|
|
|
|
return {"name": name, "version": version, "profile": profile, "models": models, "sources": sources,
|
|
"dag": {"nodes": nodes, "edges": edges}}
|
|
|
|
|
|
# ── Scan service (panel-driven) ──────────────────────────────────────────────
|
|
# Panel mengirim katalog TIPE service (regex + port); agent memindai container
|
|
# yang SEDANG BERJALAN di host ini (via docker-proxy internal) dan mengembalikan
|
|
# instance + URL-nya. Logika pengenalan tetap satu tempat di panel (TS) — agent
|
|
# hanya mesin generik. Hanya container `running` yang dipindai, sehingga job
|
|
# sekali-jalan (init/migrate) otomatis tidak ikut.
|
|
|
|
# Label peta port → skema URL asli (label boleh deskriptif, mis. "console").
|
|
SCHEME_ALIASES = {
|
|
"https": "https",
|
|
"http": "http",
|
|
"api": "http",
|
|
"console": "http",
|
|
"ui": "http",
|
|
"web": "http",
|
|
"ldap": "ldap",
|
|
"mysql": "mysql",
|
|
"sql": "https",
|
|
"tcp": "tcp",
|
|
}
|
|
def _re_match(pattern, value):
|
|
if not pattern:
|
|
return False
|
|
try:
|
|
return re.search(pattern, value or "") is not None
|
|
except re.error:
|
|
return False
|
|
|
|
|
|
def docker_running_containers():
|
|
"""Daftar container running dari docker-proxy internal."""
|
|
req = urllib.request.Request(f"{DOCKER_PROXY_URL}/containers/json?all=false")
|
|
with urllib.request.urlopen(req, timeout=8) as resp:
|
|
data = json.loads(resp.read().decode("utf-8"))
|
|
out = []
|
|
for c in data:
|
|
if (c.get("State") or "") != "running":
|
|
continue
|
|
names = c.get("Names") or []
|
|
name = names[0].lstrip("/") if names else (c.get("Id") or "")[:12]
|
|
out.append({
|
|
"name": name,
|
|
"image": c.get("Image", ""),
|
|
"labels": c.get("Labels") or {},
|
|
"ports": c.get("Ports") or [],
|
|
})
|
|
return out
|
|
|
|
|
|
def _published_ports(ports_raw):
|
|
out = []
|
|
seen = set()
|
|
for p in ports_raw:
|
|
if not p.get("PublicPort"):
|
|
continue
|
|
key = (p.get("PrivatePort"), p.get("PublicPort"), p.get("Type", "tcp"))
|
|
if key in seen:
|
|
continue # docker mengembalikan entri IPv4 & IPv6 → dedupe
|
|
seen.add(key)
|
|
out.append({
|
|
"privatePort": p.get("PrivatePort"),
|
|
"publicPort": p.get("PublicPort"),
|
|
"type": p.get("Type", "tcp"),
|
|
})
|
|
return out
|
|
|
|
|
|
def _urls_for(ports_map, published_raw, public_host):
|
|
"""Susun URL per label dari peta port container → port publish host.
|
|
|
|
Kunci peta = LABEL (mis. "http", "console", "mysql"); skema URL sebenarnya
|
|
dipetakan lewat SCHEME_ALIASES (label "console" → skema "http").
|
|
"""
|
|
urls = {}
|
|
if not public_host:
|
|
return urls
|
|
for label, privates in (ports_map or {}).items():
|
|
if not isinstance(privates, list):
|
|
continue
|
|
scheme = SCHEME_ALIASES.get(str(label).lower(), "http")
|
|
for priv in privates:
|
|
for p in published_raw:
|
|
if int(p.get("PrivatePort") or 0) == int(priv) and p.get("PublicPort"):
|
|
urls[label] = f"{scheme}://{public_host}:{p.get('PublicPort')}"
|
|
break
|
|
if label in urls:
|
|
break
|
|
return urls
|
|
|
|
|
|
def scan_services(types, public_host):
|
|
containers = docker_running_containers()
|
|
services = []
|
|
for t in types:
|
|
if not isinstance(t, dict):
|
|
continue
|
|
img_pat = t.get("imagePattern")
|
|
name_pat = t.get("namePattern")
|
|
ports_map = t.get("ports") or {}
|
|
for c in containers:
|
|
if not (_re_match(img_pat, c["image"]) or _re_match(name_pat, c["name"])):
|
|
continue
|
|
services.append({
|
|
"type": t.get("id"),
|
|
"container": c["name"],
|
|
"image": c["image"],
|
|
"labels": c["labels"],
|
|
"published": _published_ports(c["ports"]),
|
|
"urls": _urls_for(ports_map, c["ports"], public_host),
|
|
})
|
|
return {
|
|
"hostname": socket.gethostname(),
|
|
"containers": len(containers),
|
|
"config": {
|
|
"envFile": os.path.exists(ENV_FILE),
|
|
"catalogDir": os.path.isdir(CATALOG_DIR),
|
|
"dbtDir": os.path.isdir(DBT_DIR),
|
|
"caFile": bool(CA_FILE) and os.path.exists(CA_FILE),
|
|
},
|
|
"services": services,
|
|
}
|
|
|
|
|
|
# ── HTTP handler ─────────────────────────────────────────────────────────────
|
|
class Handler(BaseHTTPRequestHandler):
|
|
server_version = "sicabe-config-agent/0.1"
|
|
|
|
# -- util --
|
|
def _send(self, code, body, ctype="application/json"):
|
|
payload = body if isinstance(body, bytes) else json.dumps(body).encode("utf-8")
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", ctype)
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
try:
|
|
self.wfile.write(payload)
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
pass
|
|
|
|
def _auth_ok(self):
|
|
key = self.headers.get("X-Api-Key", "")
|
|
return bool(AGENT_API_KEY) and hmac.compare_digest(key, AGENT_API_KEY)
|
|
|
|
def _deny(self):
|
|
self._send(401, {"error": "Unauthorized"})
|
|
|
|
def _json_body(self):
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
raw = self.rfile.read(length) if length else b"{}"
|
|
return json.loads(raw.decode("utf-8"))
|
|
|
|
def log_message(self, fmt, *args): # ringkas ke stderr
|
|
print("[agent] " + (fmt % args), flush=True)
|
|
|
|
# -- GET --
|
|
def do_GET(self):
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path.rstrip("/") or "/"
|
|
|
|
if path == "/healthz":
|
|
return self._send(200, {"status": "ok"})
|
|
if not self._auth_ok():
|
|
return self._deny()
|
|
|
|
try:
|
|
if path == "/env":
|
|
keys = parse_qs(parsed.query).get("key", [])
|
|
keys = [k for k in keys if k]
|
|
if not keys:
|
|
return self._send(400, {"error": "key parameter required (whitelist)"})
|
|
return self._send(200, read_env(keys))
|
|
|
|
if path == "/catalog":
|
|
return self._send(200, {"catalogs": list_catalogs()})
|
|
|
|
if path.startswith("/catalog/"):
|
|
name = path[len("/catalog/"):]
|
|
cat = get_catalog(name)
|
|
if not cat:
|
|
return self._send(404, {"error": f"catalog '{name}' not found"})
|
|
return self._send(200, cat)
|
|
|
|
if path == "/dbt":
|
|
return self._send(200, load_dbt())
|
|
|
|
if path == "/ca":
|
|
if not CA_FILE or not os.path.exists(CA_FILE):
|
|
return self._send(404, {"error": "no CA configured"})
|
|
# Kirim bytes mentah (jangan di-JSON-encode) agar PEM utuh.
|
|
with open(CA_FILE, "rb") as fh:
|
|
return self._send(200, fh.read(), "application/x-pem-file")
|
|
|
|
if path == "/opa/data":
|
|
# Kirim isi file APA ADANYA. Harus bytes: _send() men-JSON-encode
|
|
# nilai non-bytes (kalau str, isi file jadi string ber-quote ganda
|
|
# dan editor panel menampilkan JSON yang salah).
|
|
if not os.path.exists(OPA_DATA_FILE):
|
|
return self._send(404, {"error": "opa data file not found"})
|
|
with open(OPA_DATA_FILE, "rb") as fh:
|
|
return self._send(200, fh.read(), "application/json")
|
|
|
|
return self._send(404, {"error": "not found"})
|
|
except Exception as e: # noqa: BLE001
|
|
return self._send(500, {"error": "internal error", "details": str(e)})
|
|
|
|
# -- PUT --
|
|
def do_PUT(self):
|
|
if not self._auth_ok():
|
|
return self._deny()
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path.rstrip("/") or "/"
|
|
|
|
if path == "/opa/data":
|
|
try:
|
|
body = self._json_body()
|
|
content = body.get("content") if isinstance(body, dict) else None
|
|
if not isinstance(content, str):
|
|
return self._send(400, {"error": "content (JSON string) required"})
|
|
# Validasi SEBELUM menulis: JSON invalid membuat OPA gagal start
|
|
# → authz Trino mati total.
|
|
try:
|
|
json.loads(content)
|
|
except json.JSONDecodeError as e:
|
|
return self._send(400, {"error": "invalid JSON", "details": e.msg,
|
|
"line": e.lineno, "col": e.colno})
|
|
write_opa_data(content)
|
|
return self._send(200, {"success": True, "bytes": len(content.encode("utf-8"))})
|
|
except Exception as e: # noqa: BLE001
|
|
return self._send(500, {"error": "internal error", "details": str(e)})
|
|
|
|
if not path.startswith("/catalog/"):
|
|
return self._send(404, {"error": "not found"})
|
|
name = path[len("/catalog/"):]
|
|
if not VALID_CATALOG.match(name):
|
|
return self._send(400, {"error": "invalid catalog name"})
|
|
try:
|
|
body = self._json_body()
|
|
props = body.get("properties", body) if isinstance(body, dict) else {}
|
|
if not isinstance(props, dict):
|
|
return self._send(400, {"error": "properties must be an object"})
|
|
put_catalog(name, {str(k): str(v) for k, v in props.items()})
|
|
return self._send(200, {"success": True, "name": name})
|
|
except Exception as e: # noqa: BLE001
|
|
return self._send(500, {"error": "internal error", "details": str(e)})
|
|
|
|
# -- DELETE --
|
|
def do_DELETE(self):
|
|
if not self._auth_ok():
|
|
return self._deny()
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path.rstrip("/") or "/"
|
|
if not path.startswith("/catalog/"):
|
|
return self._send(404, {"error": "not found"})
|
|
name = path[len("/catalog/"):]
|
|
if not VALID_CATALOG.match(name):
|
|
return self._send(400, {"error": "invalid catalog name"})
|
|
try:
|
|
delete_catalog(name)
|
|
return self._send(200, {"success": True, "name": name})
|
|
except FileNotFoundError:
|
|
return self._send(404, {"error": f"catalog '{name}' not found"})
|
|
except Exception as e: # noqa: BLE001
|
|
return self._send(500, {"error": "internal error", "details": str(e)})
|
|
|
|
|
|
# -- POST --
|
|
def do_POST(self):
|
|
if not self._auth_ok():
|
|
return self._deny()
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path.rstrip("/") or "/"
|
|
|
|
if path == "/services/scan":
|
|
try:
|
|
body = self._json_body()
|
|
types = body.get("types") if isinstance(body, dict) else None
|
|
if not isinstance(types, list) or not types:
|
|
return self._send(400, {"error": "types (non-empty array) required"})
|
|
public_host = body.get("publicHost") or ""
|
|
return self._send(200, scan_services(types, public_host))
|
|
except Exception as e: # noqa: BLE001
|
|
return self._send(500, {"error": "scan failed", "details": str(e)})
|
|
|
|
return self._send(404, {"error": "not found"})
|
|
|
|
|
|
def main():
|
|
if not AGENT_API_KEY:
|
|
print("FATAL: AGENT_API_KEY must be set", flush=True)
|
|
raise SystemExit(1)
|
|
print(f"[agent] listening on :{AGENT_PORT} (env={ENV_FILE}, catalog={CATALOG_DIR}, dbt={DBT_DIR})", flush=True)
|
|
ThreadingHTTPServer(("0.0.0.0", AGENT_PORT), Handler).serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |