diff --git a/agent-workspace/agent_helpers.py b/agent-workspace/agent_helpers.py index 2d493c17..8b1858a1 100644 --- a/agent-workspace/agent_helpers.py +++ b/agent-workspace/agent_helpers.py @@ -5,3 +5,359 @@ repo's default agent-workspace exists. """ +from __future__ import annotations + +import os +import re +from contextlib import contextmanager +from pathlib import Path +from typing import Any, Iterator, Optional +from urllib.parse import urlparse + + +# --- Invicti Enterprise API helpers (netsparkercloud.com) --- +# These helpers are intentionally read-oriented: they make it easy for an agent to +# answer questions about websites, schedules, vulnerabilities/issues, etc. + +_INVICTI_DEFAULT_BASE_URL = "https://www.netsparkercloud.com" + + +def _invicti_repo_root() -> Path: + # agent-workspace/agent_helpers.py -> repo_root/agent-workspace + return Path(__file__).resolve().parents[1] + + +def _invicti_maybe_load_env() -> None: + """Best-effort load of INVICTI_* variables from common .env locations. + + browser_harness.helpers already loads REPO_ROOT/.env and agent-workspace/.env. + This additionally supports a domain-scoped env file at domain-skills/invicti/.env. + """ + + if os.environ.get("INVICTI_USER_ID") and os.environ.get("INVICTI_TOKEN"): + return + + candidates = [ + _invicti_repo_root() / ".env", + _invicti_repo_root() / "agent-workspace" / ".env", + _invicti_repo_root() / "domain-skills" / "invicti" / ".env", + ] + wanted = {"INVICTI_BASE_URL", "INVICTI_USER_ID", "INVICTI_TOKEN"} + + for p in candidates: + if not p.exists(): + continue + for line in p.read_text().splitlines(): + 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().strip('"').strip("'") + if v: + os.environ.setdefault(k, v) + + +def invicti_creds() -> tuple[str, str, str]: + """Return (base_url, user_id, token). Raises if missing.""" + + _invicti_maybe_load_env() + + base = (os.environ.get("INVICTI_BASE_URL") or _INVICTI_DEFAULT_BASE_URL).rstrip("/") + uid = (os.environ.get("INVICTI_USER_ID") or "").strip() + tok = (os.environ.get("INVICTI_TOKEN") or "").strip() + + if not uid or not tok: + raise RuntimeError( + "Missing Invicti API credentials. Set INVICTI_USER_ID and INVICTI_TOKEN " + "(recommended: repo .env or agent-workspace/.env)." + ) + + return base, uid, tok + + +@contextmanager +def invicti_client(timeout: float = 60.0) -> Iterator["httpx.Client"]: + """Context-managed httpx client authenticated with Invicti Basic Auth.""" + + import httpx + + base, uid, tok = invicti_creds() + with httpx.Client(base_url=base, auth=(uid, tok), timeout=timeout) as c: + yield c + + +def invicti_request( + method: str, + path: str, + *, + params: Optional[dict[str, Any]] = None, + json: Any = None, + timeout: float = 60.0, + allow_404: bool = False, +) -> Any: + """Call the Invicti API. + + - Returns decoded JSON when content-type is application/json. + - Returns text otherwise. + - If allow_404=True and the API returns 404, returns None. + """ + + method = method.upper().strip() + + with invicti_client(timeout=timeout) as c: + r = c.request(method, path, params=params, json=json) + if allow_404 and r.status_code == 404: + return None + r.raise_for_status() + + ct = (r.headers.get("content-type") or "").split(";", 1)[0].strip().lower() + if ct == "application/json": + return r.json() + return r.text + + +def invicti_get( + path: str, + params: Optional[dict[str, Any]] = None, + *, + timeout: float = 60.0, + allow_404: bool = False, +) -> Any: + return invicti_request("GET", path, params=params, timeout=timeout, allow_404=allow_404) + + +def invicti_paged_list( + path: str, + params: Optional[dict[str, Any]] = None, + *, + page_size: int = 200, + max_pages: Optional[int] = None, + timeout: float = 60.0, +) -> tuple[list[Any], dict[str, Any]]: + """Fetch a list endpoint. + + Supports both shapes: + - Paged wrapper dict with a "List" field + - Raw JSON arrays + + max_pages limits the number of pages fetched (not the page index). + """ + + items: list[Any] = [] + meta: dict[str, Any] = {} + + p0 = dict(params or {}) + page = int(p0.get("page") or 1) + pages_fetched = 0 + + while True: + pages_fetched += 1 + + p = dict(params or {}) + p["page"] = page + p.setdefault("pageSize", page_size) + + data = invicti_get(path, params=p, timeout=timeout) + + if isinstance(data, dict) and "List" in data: + items.extend(data.get("List") or []) + meta = { + k: data.get(k) + for k in ( + "TotalItemCount", + "PageNumber", + "PageSize", + "PageCount", + "HasNextPage", + "HasPreviousPage", + ) + if k in data + } + + if not data.get("HasNextPage"): + break + if max_pages is not None and pages_fetched >= max_pages: + break + + page += 1 + continue + + if isinstance(data, list): + items = data + meta = {"TotalItemCount": len(items), "nonPaged": True} + break + + meta = {"shape": type(data).__name__} + break + + return items, meta + + +def invicti_website_get(query: str) -> Optional[dict[str, Any]]: + """Get a website by name or URL. Returns None if not found.""" + + data = invicti_get("/api/1.0/websites/get", params={"query": query}, allow_404=True) + return data if isinstance(data, dict) else None + + +def invicti_website_search(search_term: str, *, page_size: int = 20) -> list[dict[str, Any]]: + """Search websites by term (first page only).""" + + items, _ = invicti_paged_list( + "/api/1.0/websites/searchlist", + params={"searchTerm": search_term}, + page_size=page_size, + max_pages=1, + ) + return [x for x in items if isinstance(x, dict)] + + +def invicti_resolve_website(query: str, *, max_results: int = 10) -> list[dict[str, Any]]: + """Return 0..N candidate website dicts for a name/URL query.""" + + w = invicti_website_get(query) + if w: + return [w] + + term = query.strip() + + # If URL-ish, search by host first (often more stable). + try: + u = term if re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*://", term) else "https://" + term + host = urlparse(u).netloc + if host: + term = host + except Exception: + pass + + return invicti_website_search(term, page_size=max_results)[:max_results] + + +def _invicti_norm_urlish(value: str) -> str: + """Normalize URL-ish strings for comparisons (ignore scheme, trim trailing '/').""" + + if not value: + return "" + + v = value.strip() + if not re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*://", v): + v = "https://" + v + + p = urlparse(v) + host = (p.netloc or "").lower() + path = (p.path or "").rstrip("/") + return f"{host}{path}" + + +def invicti_list_scheduled_scans( + *, + page_size: int = 200, + max_pages: Optional[int] = None, +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + items, meta = invicti_paged_list( + "/api/1.0/scans/list-scheduled", + page_size=page_size, + max_pages=max_pages, + ) + return [x for x in items if isinstance(x, dict)], meta + + +def invicti_schedules_for_website(query: str) -> list[dict[str, Any]]: + """Return scheduled scan entries whose TargetUrl matches a website query.""" + + candidates = {_invicti_norm_urlish(query)} + + for w in invicti_resolve_website(query, max_results=3): + candidates.add(_invicti_norm_urlish(w.get("RootUrl") or "")) + candidates.add(_invicti_norm_urlish(w.get("Name") or "")) + + candidates = {c for c in candidates if c} + + schedules, _ = invicti_list_scheduled_scans(page_size=200, max_pages=None) + + out: list[dict[str, Any]] = [] + for s in schedules: + targets: list[str] = [] + + if isinstance(s.get("TargetUrl"), str) and s.get("TargetUrl"): + targets.append(s["TargetUrl"]) + + addl = s.get("AdditionalWebsites") + if isinstance(addl, list): + for x in addl: + if isinstance(x, str): + targets.append(x) + elif isinstance(x, dict): + for k in ("TargetUrl", "Url", "RootUrl"): + if isinstance(x.get(k), str) and x.get(k): + targets.append(x[k]) + break + + if any(_invicti_norm_urlish(t) in candidates for t in targets if t): + out.append(s) + + return out + + +def invicti_issue_counts( + website_name: str, + *, + severities: Optional[list[str]] = None, +) -> dict[str, int]: + """Return TotalItemCount per severity for /issues/allissues (rawDetails=false).""" + + severities = severities or [ + "Critical", + "High", + "Medium", + "Low", + "Information", + "BestPractice", + ] + + counts: dict[str, int] = {} + for sev in severities: + data = invicti_get( + "/api/1.0/issues/allissues", + params={ + "webSiteName": website_name, + "severity": sev, + "page": 1, + "pageSize": 1, + "rawDetails": False, + }, + ) + total = data.get("TotalItemCount") if isinstance(data, dict) else None + counts[sev] = int(total or 0) + + return counts + + +def invicti_issues_sample( + website_name: str, + *, + severity: str = "Critical", + limit: int = 50, +) -> list[dict[str, Any]]: + """Fetch up to `limit` issues for a website+severity (rawDetails=false).""" + + page_size = min(200, max(1, limit)) + max_pages = (limit + page_size - 1) // page_size + + items, _ = invicti_paged_list( + "/api/1.0/issues/allissues", + params={ + "webSiteName": website_name, + "severity": severity, + "rawDetails": False, + }, + page_size=page_size, + max_pages=max_pages, + ) + + out = [x for x in items if isinstance(x, dict)] + return out[:limit] + diff --git a/domain-skills/invicti/invicit-api/SKILL.md b/domain-skills/invicti/invicit-api/SKILL.md new file mode 100644 index 00000000..c709024d --- /dev/null +++ b/domain-skills/invicti/invicit-api/SKILL.md @@ -0,0 +1,124 @@ +--- +name: invicti-api +description: Query and reason about Invicti Enterprise (Netsparker Cloud) via its Web API (netsparkercloud.com). Use whenever the user mentions Invicti/Netsparker, scan schedules, scheduled scans, scan profiles, websites, website groups, vulnerabilities/issues, technologies, agents, teams/roles/members, or asks questions like “Schedules”, “does website X have a schedule?”, or “does website X have vulnerabilities?”. +--- + +# Invicti Enterprise API (Netsparker Cloud) +Use the documented Invicti Enterprise API: +- Docs UI: https://www.netsparkercloud.com/docs/index +- Swagger JSON: https://www.netsparkercloud.com/swagger/docs/v1 +- Base path for API calls: `https://www.netsparkercloud.com/api/1.0` + +## Authentication (required) +The API uses HTTP Basic Auth: +- username: `INVICTI_USER_ID` +- password: `INVICTI_TOKEN` + +Prefer setting these as environment variables (or in a local `.env`, which this repo already ignores): +- `INVICTI_BASE_URL` (optional; default `https://www.netsparkercloud.com`) +- `INVICTI_USER_ID` +- `INVICTI_TOKEN` + +Do not print or commit credentials. + +## Read-only first (avoid surprises) +Default to GET/read-only endpoints. +If the user asks for a write action (schedule/unschedule scans, update issues, create websites, etc.), ask for explicit confirmation before sending POST/PUT/DELETE. + +## Built-in helper functions (recommended) +If you are operating inside this `browser-harness` repo, prefer the Invicti helpers in `agent-workspace/agent_helpers.py`: +- `invicti_get(path, params=...)` +- `invicti_paged_list(path, params=..., page_size=..., max_pages=...)` +- `invicti_resolve_website(query)` +- `invicti_list_scheduled_scans()` / `invicti_schedules_for_website(query)` +- `invicti_issue_counts(website_name)` / `invicti_issues_sample(website_name, ...)` + +These helpers intentionally avoid printing secrets and default to `rawDetails=false` for issues. + +## How to answer common user questions +### “Schedules” / “List scheduled scans” +Use: +- `GET /api/1.0/scans/list-scheduled?page=&pageSize=` + +Return a compact list with (at minimum): +- `Name` +- `TargetUrl` +- `NextExecutionTime` +- `ScheduleRunType` +- `Disabled` / `EnableScheduling` + +Note: `list-scheduled` returns *scheduled scan entries* (often scan-profile-like objects). Filter client-side. + +### “Does website have a schedule?” +Steps: +1. Resolve the website (accept URL, root URL, or display name): + - `GET /api/1.0/websites/get?query=` + - If not found, fallback to `GET /api/1.0/websites/searchlist?searchTerm=` and pick the best match. +2. List scheduled scans. +3. Match schedules to the website primarily by URL: + - Compare `schedule.TargetUrl` to `website.RootUrl` (normalize trailing `/`, ignore scheme differences when needed). + - Also check `schedule.AdditionalWebsites` when present. +4. Report: + - whether a schedule exists + - which schedule(s) matched + - next run time(s) + +### “Does website have vulnerabilities?” +Use the Issues API with the website name: +- `GET /api/1.0/issues/allissues?webSiteName=&severity=<...>&page=&pageSize=&rawDetails=false` + +Practical pattern: +- Query `TotalItemCount` across severities (`Critical`, `High`, `Medium`, `Low`, `Information`, `BestPractice`). +- If all are 0, the website has no issues recorded for those severities. +- If non-zero, optionally fetch a small sample and group by: + - `Type` (stable, good for dedup) + - `Title` (human readable) + +Avoid `rawDetails=true` unless the user explicitly asks for proof-of-concept / request-response payloads. + +### “What vulnerabilities does website have?” +Use the same Issues endpoint, but pull a bounded sample (or multiple pages when requested) and summarize: +- counts by severity +- top `Type` / `Title` +- last seen / first seen dates for a few representative findings + +### “Technologies for website ” +Use: +- `GET /api/1.0/technologies/list?webSiteName=&page=&pageSize=` + +Summarize: +- total technology records +- number out-of-date (`IsOutofDate`) +- top tech by critical/high issue counts + +## Endpoint quick reference +Websites: +- `GET /api/1.0/websites/get?query=...` +- `GET /api/1.0/websites/searchlist?searchTerm=...` +- `GET /api/1.0/websites/list?page=&pageSize=` +- `GET /api/1.0/websites/get/{id}` + +Schedules (scheduled scans): +- `GET /api/1.0/scans/list-scheduled?page=&pageSize=` +- `GET /api/1.0/scans/get-scheduled/{id}` + +Issues (vulnerabilities): +- `GET /api/1.0/issues/allissues` (filter by `webSiteName`, `severity`) +- `GET /api/1.0/issues/todo` +- `GET /api/1.0/issues/addressedissues` +- `GET /api/1.0/issues/report` (CSV export) + +Stacks: +- `GET /api/1.0/technologies/list` +- `GET /api/1.0/vulnerability/list` (definitions/templates) +- `GET /api/1.0/vulnerability/types` (type strings) + +## Swagger discovery workflow (when you need a new endpoint) +When unsure which endpoint supports a feature, search the Swagger JSON for tags/paths/keywords: +- https://www.netsparkercloud.com/swagger/docs/v1 + +Example approach: +1. Fetch Swagger JSON. +2. Filter paths by keyword (e.g., `schedule`, `scan`, `issues`, `website`). +3. Read `parameters` to learn required query/body fields. +4. Prefer endpoints that support server-side filtering; otherwise page and filter locally. diff --git a/domain-skills/invicti/invicti-api-verification.md b/domain-skills/invicti/invicti-api-verification.md new file mode 100644 index 00000000..849b61ba --- /dev/null +++ b/domain-skills/invicti/invicti-api-verification.md @@ -0,0 +1,35 @@ +# Invicti API verification (read-only) + +Open the docs UI and authenticate: +- https://www.netsparkercloud.com/docs/index + +Credentials are required (HTTP Basic Auth) and should be provided via local env vars / `.env`: +- `INVICTI_USER_ID` +- `INVICTI_TOKEN` + +Then produce a summary of the following datasets (read-only endpoints): + +## Humans +- Current user: `GET /api/1.0/account/me` +- Teams: `GET /api/1.0/team/list` +- Roles: `GET /api/1.0/roles/list` +- Members: `GET /api/1.0/members/list` + +## Groups +- AgentGroups: `GET /api/1.0/agentgroups/list` +- Agents: `GET /api/1.0/agents/list` +- WebsiteGroups: `GET /api/1.0/websitegroups/list` +- Websites: `GET /api/1.0/websites/list` (or use `GET /api/1.0/websites/searchlist` / `GET /api/1.0/websites/get` for targeted lookups) + +## Stacks +- Technologies: `GET /api/1.0/technologies/list` +- Vulnerability definitions: `GET /api/1.0/vulnerability/list` +- Vulnerability types: `GET /api/1.0/vulnerability/types` + +## Correlation ideas +- Join `Members[].RoleWebsiteGroupMappings[]` and `Team[].RoleWebsiteGroupMappings[]` to WebsiteGroups by `WebsiteGroupId` to reason about access scope and website counts. +- Join `AgentGroups[].Agents[]` (agent IDs) to `Agents[].Id`. +- Websites include `Groups` (id+name) which can be correlated to WebsiteGroups. + +Optional: export separate CSV files (Members, Teams, Roles, AgentGroups, Agents, WebsiteGroups, Websites, Technologies) and pivot locally. + diff --git a/domain-skills/invicti/invicti-api-verification.png b/domain-skills/invicti/invicti-api-verification.png new file mode 100644 index 00000000..07ad2020 Binary files /dev/null and b/domain-skills/invicti/invicti-api-verification.png differ