83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
import os
|
|
|
|
import requests
|
|
import urllib3
|
|
|
|
from ...models import DesiredRecord
|
|
from ...normalizer import normalize_hostname
|
|
|
|
|
|
class Plugin:
|
|
name = "npm"
|
|
|
|
def __init__(self, config):
|
|
self.config = config
|
|
self.identity = os.getenv("NPM_IDENTITY")
|
|
self.secret = os.getenv("NPM_SECRET")
|
|
if not self.identity:
|
|
raise RuntimeError("NPM_IDENTITY is required when PROXY=npm")
|
|
if not self.secret:
|
|
raise RuntimeError("NPM_SECRET is required when PROXY=npm")
|
|
|
|
self.session = requests.Session()
|
|
self.token = None
|
|
if not config.proxy_verify_tls:
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
@property
|
|
def api_url(self):
|
|
return f"{self.config.proxy_url}/api"
|
|
|
|
def _login(self):
|
|
response = self.session.post(
|
|
f"{self.api_url}/tokens",
|
|
json={"identity": self.identity, "secret": self.secret},
|
|
timeout=15,
|
|
verify=self.config.proxy_verify_tls,
|
|
)
|
|
if response.status_code != 200:
|
|
raise RuntimeError(f"NPM login failed: HTTP {response.status_code}: {response.text[:300]}")
|
|
data = response.json()
|
|
token = data.get("token")
|
|
if not token:
|
|
raise RuntimeError("NPM login succeeded but no token was returned")
|
|
self.token = token
|
|
self.session.headers.update({"Authorization": f"Bearer {token}", "Accept": "application/json"})
|
|
|
|
def _proxy_hosts(self):
|
|
if not self.token:
|
|
self._login()
|
|
response = self.session.get(
|
|
f"{self.api_url}/nginx/proxy-hosts",
|
|
timeout=15,
|
|
verify=self.config.proxy_verify_tls,
|
|
)
|
|
if response.status_code != 200:
|
|
raise RuntimeError(f"Unable to retrieve NPM proxy hosts: HTTP {response.status_code}: {response.text[:300]}")
|
|
data = response.json()
|
|
if not isinstance(data, list):
|
|
raise RuntimeError("NPM proxy-host response was not a list")
|
|
return data
|
|
|
|
def discover(self):
|
|
records: list[DesiredRecord] = []
|
|
for host in self._proxy_hosts():
|
|
enabled = bool(host.get("enabled", False))
|
|
if not enabled and not self.config.include_disabled:
|
|
continue
|
|
|
|
source_id = str(host.get("id", "")) or None
|
|
for raw_domain in host.get("domain_names", []):
|
|
hostname, wildcard = normalize_hostname(str(raw_domain), allow_wildcard=True)
|
|
records.append(
|
|
DesiredRecord(
|
|
hostname=hostname,
|
|
ip=self.config.target_ip,
|
|
record_type="A",
|
|
wildcard=wildcard,
|
|
source=self.name,
|
|
source_id=source_id,
|
|
)
|
|
)
|
|
return records
|