diff --git a/docs/advanced/skills.mdx b/docs/advanced/skills.mdx index 8d9ced96..ce624424 100644 --- a/docs/advanced/skills.mdx +++ b/docs/advanced/skills.mdx @@ -68,10 +68,10 @@ Framework-specific testing patterns. Third-party service and platform security. -| Skill | Coverage | -| -------------------- | ---------------------------------- | -| `supabase` | Supabase RLS bypasses, auth issues | -| `firebase_firestore` | Firestore rules, Firebase auth | +| Skill | Coverage | +| ---------- | ------------------------------------------------------ | +| `supabase` | Supabase RLS bypasses, auth issues | +| `firebase` | Firebase Firestore, Storage rules, Auth, and Functions | ### Protocols diff --git a/strix/agents/prompts/system_prompt.jinja b/strix/agents/prompts/system_prompt.jinja index 6af47c4e..5fc697d7 100644 --- a/strix/agents/prompts/system_prompt.jinja +++ b/strix/agents/prompts/system_prompt.jinja @@ -490,8 +490,10 @@ Default user: pentester (sudo available) On-demand specialist skills. Spawn a specialist via `create_agent(skills=[...])`, or pull guidance inline for yourself via `load_skill(skills=[...])`. Anything wrapped in `` above is already loaded for you. -{% for category, names in available_skills | dictsort -%} -- {{ category }}: {{ names | join(', ') }} +{% for category, skills in available_skills | dictsort -%} +{% for skill in skills -%} +- {{ category }}/{{ skill.name }}{% if skill.description %}: {{ skill.description }}{% endif %} +{% endfor -%} {% endfor -%} {% endif %} diff --git a/strix/skills/__init__.py b/strix/skills/__init__.py index 31ac3059..0adf3d99 100644 --- a/strix/skills/__init__.py +++ b/strix/skills/__init__.py @@ -4,6 +4,9 @@ import threading from collections import Counter from collections.abc import Iterator from pathlib import Path +from typing import TypeGuard + +import yaml from strix.telemetry import posthog, scarf from strix.utils.resource_paths import get_strix_resource_path @@ -11,12 +14,17 @@ from strix.utils.resource_paths import get_strix_resource_path logger = logging.getLogger(__name__) -_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n.*?\n---\s*\n", re.DOTALL) +_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n(?P.*?)\n---\s*\n", re.DOTALL) _INTERNAL_SKILL_CATEGORIES: frozenset[str] = frozenset({"scan_modes", "coordination"}) _ROOT_SKILL_CATEGORY = "root" _EXTRA_SKILL_DIRS: list[Path] = [] +_SKILL_METADATA_CACHE: dict[tuple[Path, int, int], dict[str, str]] = {} + + +def _is_frontmatter_mapping(value: object) -> TypeGuard[dict[object, object]]: + return isinstance(value, dict) def register_skill_dir(path: str | Path) -> None: @@ -109,13 +117,18 @@ def _get_ambiguous_skill_names() -> set[str]: return {name for name, count in counts.items() if count > 1} -def _qualified_skill_files(skill_name: str) -> list[Path]: +def _qualified_skill_file_for_name(skill_name: str) -> Path | None: category, _, name = skill_name.partition("/") for skills_dir in skill_search_dirs(): candidate = _qualified_skill_file(skills_dir, category, name) if candidate is not None: - return [candidate] - return [] + return candidate + return None + + +def _qualified_skill_files(skill_name: str) -> list[Path]: + candidate = _qualified_skill_file_for_name(skill_name) + return [candidate] if candidate is not None else [] def _bare_skill_files(skill_name: str) -> list[Path]: @@ -145,10 +158,59 @@ def _bare_skill_files(skill_name: str) -> list[Path]: return candidates -def get_available_skills() -> dict[str, list[str]]: - grouped: dict[str, list[str]] = {} +def _parse_skill_content(content: str, source: Path | None = None) -> tuple[dict[str, str], str]: + """Parse skill frontmatter once and return metadata plus markdown body.""" + frontmatter = _FRONTMATTER_PATTERN.match(content) + if frontmatter is None: + return {}, content.lstrip() + + try: + parsed: object = yaml.safe_load(frontmatter.group("body")) + except yaml.YAMLError as error: + logger.warning("Failed to parse skill frontmatter %s: %s", source or "", error) + parsed = None + if not _is_frontmatter_mapping(parsed): + logger.warning("Skill frontmatter is not a mapping: %s", source or "") + return {}, content[frontmatter.end() :].lstrip() + + metadata = {str(key): "" if value is None else str(value) for key, value in parsed.items()} + return metadata, content[frontmatter.end() :].lstrip() + + +def _read_skill_metadata(file_path: Path) -> dict[str, str]: + try: + stat = file_path.stat() + except OSError: + logger.warning("Skill file disappeared while reading metadata: %s", file_path) + return {} + cache_key = (file_path, stat.st_mtime_ns, stat.st_size) + cached = _SKILL_METADATA_CACHE.get(cache_key) + if cached is not None: + return cached + try: + content = file_path.read_text(encoding="utf-8") + except (OSError, ValueError): + logger.warning("Failed to read skill metadata: %s", file_path) + return {} + metadata, _ = _parse_skill_content(content, file_path) + _SKILL_METADATA_CACHE[cache_key] = metadata + return metadata + + +def get_available_skills() -> dict[str, list[dict[str, str]]]: + grouped: dict[str, list[dict[str, str]]] = {} for category, name in _iter_user_skill_files(): - grouped.setdefault(category, []).append(name) + file_path = _qualified_skill_file_for_name(f"{category}/{name}") + if file_path is None: + logger.warning( + "Skill disappeared while gathering available skills: %s/%s", + category, + name, + ) + continue + metadata = _read_skill_metadata(file_path) + description = " ".join(metadata.get("description", "").split()) + grouped.setdefault(category, []).append({"name": name, "description": description}) return grouped @@ -228,7 +290,8 @@ def load_skills(skill_names: list[str]) -> dict[str, str]: continue var_name = skill_name.split("/")[-1] - skill_content[var_name] = _FRONTMATTER_PATTERN.sub("", content).lstrip() + _, skill_body = _parse_skill_content(content, file_path) + skill_content[var_name] = skill_body logger.debug("Loaded skill: %s -> %s", skill_name, var_name) _track_skill_loaded(var_name, file_path) diff --git a/strix/skills/cloud/gcp.md b/strix/skills/cloud/gcp.md index a6f28293..2c721594 100644 --- a/strix/skills/cloud/gcp.md +++ b/strix/skills/cloud/gcp.md @@ -16,7 +16,7 @@ GCP misconfigurations expose project data, service account keys, and lateral mov **Storage & Data** - Cloud Storage (GCS) buckets and objects -- BigQuery datasets, Cloud SQL instances, Firestore (see `firebase_firestore` skill) +- BigQuery datasets, Cloud SQL instances, Firestore (see `firebase` skill) - Secret Manager, Cloud KMS keys **Compute** diff --git a/strix/skills/technologies/firebase_firestore.md b/strix/skills/technologies/firebase.md similarity index 64% rename from strix/skills/technologies/firebase_firestore.md rename to strix/skills/technologies/firebase.md index a0c48728..34104033 100644 --- a/strix/skills/technologies/firebase_firestore.md +++ b/strix/skills/technologies/firebase.md @@ -1,9 +1,9 @@ --- -name: firebase-firestore -description: Firebase/Firestore security testing covering security rules, Cloud Functions, and client-side trust issues +name: firebase +description: Firebase security testing covering Firestore, Storage rules, Realtime Database, Auth, Functions, and client-side trust issues --- -# Firebase / Firestore +# Firebase Security testing for Firebase applications. Focus on Firestore/Realtime Database rules, Cloud Storage exposure, callable/onRequest Functions trusting client input, and incorrect ID token validation. @@ -30,7 +30,17 @@ Security testing for Firebase applications. Focus on Firestore/Realtime Database **Endpoints** - Firestore REST: `https://firestore.googleapis.com/v1/projects//databases/(default)/documents/` - Realtime DB: `https://.firebaseio.com/.json` -- Storage REST: `https://storage.googleapis.com/storage/v1/b/` +- GCS JSON API: `https://storage.googleapis.com/storage/v1/b/` +- Firebase Storage rules API: `https://firebasestorage.googleapis.com/v0/b//o` + +Cloud Storage has two front doors with different authorization engines: + +| Front door | Authorization engine | +| --- | --- | +| `storage.googleapis.com//` and `/storage/v1/b/` | GCS IAM and per-object ACLs | +| `firebasestorage.googleapis.com/v0/b//o` | Firebase Storage Security Rules | + +A `403` from a GCS URL does not prove that Firebase Storage rules deny access. Always test both doors. **Auth** - Google-signed ID tokens (iss: `accounts.google.com` or `securetoken.google.com/`) @@ -117,9 +127,43 @@ exists(/databases/(default)/documents/orgs/$(org)/members/$(request.auth.uid)) - Public reads on sensitive buckets/paths - Signed URLs with long TTL, no content-disposition controls, replayable across tenants - List operations exposed: `/o?prefix=` enumerates object keys +- Firebase Storage rules allowing unauthenticated or overly broad reads and writes + +**Firebase Storage rules checks** + +Probe the rules door separately from GCS IAM and ACLs: + +1. Unauthenticated list: `GET https://firebasestorage.googleapis.com/v0/b//o?prefix=` +2. Unauthenticated read of a known object path +3. Unauthenticated write/upload to a uniquely named test object +4. Repeat list, read, and write as an anonymous-auth principal when anonymous sign-in is enabled +5. Repeat the same matrix as a low-privilege authenticated user + +Write access is as important as read access and is routinely missed. Record status, response body, and object existence after each attempt; clean up only test objects that the test principal created. + +Review rules source when present and flag: + +- `allow read, write: if request.time < timestamp.date(...)` — the common console test-mode time gate +- `{allPaths=**}` catch-alls +- `request.auth != null` as the sole authorization gate +- Claim-presence checks such as `request.auth.token.roles.size() > 0` without role or tenant validation + +Storage rules use OR-across-matches semantics: a later permissive match can reopen a path that an earlier match denied. Review every matching path, not only the most specific-looking deny. + +**Bucket discovery** + +- Extract `storageBucket` from `firebase.apps[0].options` and `NEXT_PUBLIC_FIREBASE_*` values in JavaScript bundles and source. +- Check `.appspot.com` and `.firebasestorage.app` bucket conventions. + +**ACL and IAM checks are separate** + +- Sweep object ACLs for `allUsers` and `allAuthenticatedUsers`, including objects made public by Admin SDK `makePublic()` or writers using `public: true`. Per-object public ACLs persist after Firebase rules are tightened and can remain on older prefixes. +- Check bucket IAM for `allUsers` and `allAuthenticatedUsers`. +- Check whether Uniform Bucket-Level Access is disabled; legacy object ACLs matter when it is off. +- Account for CDN caching of previously public objects; cache-bust when verifying a revocation. **Tests** -- GET gs:// paths via HTTPS without auth; verify Content-Type and `Content-Disposition: attachment` +- GET GCS object paths via HTTPS without auth; verify Content-Type and `Content-Disposition: attachment` - Generate and reuse signed URLs across accounts and paths; try case/URL-encoding variants - Upload HTML/SVG and verify `X-Content-Type-Options: nosniff`; check for script execution @@ -189,12 +233,19 @@ Apps often implement multi-tenant data models (`orgs//...`). Bind tenant ## Testing Methodology -1. **Extract config** - Get project config from client bundle -2. **Obtain principals** - Collect tokens for unauth, anonymous, user A/B, admin +1. **Extract config** - Get project and storage bucket config from client bundles and source +2. **Obtain principals** - Collect tokens for unauth, anonymous, user A/B, and admin where authorized 3. **Build matrix** - Resource × Action × Principal across Firestore/Realtime/Storage/Functions -4. **SDK vs REST** - Exercise every action via both to detect parity gaps -5. **Seed IDs** - Start from list/query paths to gather document IDs -6. **Cross-principal** - Swap document paths, tenants, and user IDs across principals +4. **Exercise both Storage doors** - Test Firebase Storage rules endpoints separately from GCS IAM/ACL URLs +5. **SDK vs REST** - Exercise every action via both to detect parity gaps +6. **Seed IDs** - Start from list/query paths to gather document and object paths +7. **Cross-principal** - Swap document paths, tenants, and user IDs across principals + +## Whitebox Rules Review + +- Inspect `firebase.json`, `.firebaserc`, deployment scripts, CI configuration, and infrastructure code for `storage.rules` / `firestore.rules` declarations. +- If `firebase.json` has no `storage` or `firestore` block, or the referenced rules file is absent from the tree, treat the live rules as unmanaged and force the live probe matrix. Absence of rules IaC is itself a finding; never conclude that there is nothing to review. +- Correlate configured rule files with deployed project and bucket identifiers. A source rule file for a different project does not establish live protection. ## Tooling @@ -206,6 +257,7 @@ Apps often implement multi-tenant data models (`orgs//...`). Bind tenant ## Validation Requirements - Owner vs non-owner Firestore queries showing unauthorized access or metadata leak -- Cloud Storage read/write beyond intended scope (public object, signed URL reuse, list exposure) +- Firebase Storage unauthenticated, anonymous, or low-privilege read/list/write beyond intended scope, with minimal reproducible requests and observed deltas +- GCS object ACL or bucket IAM access beyond intended scope, including public object persistence after rules changes - Function accepting forged/foreign identity (wrong `aud`/`iss`) or trusting client `uid`/`orgId` - Minimal reproducible requests with roles/tokens used and observed deltas diff --git a/tests/test_skill_dir_extension.py b/tests/test_skill_dir_extension.py index 0c21660c..eb28768c 100644 --- a/tests/test_skill_dir_extension.py +++ b/tests/test_skill_dir_extension.py @@ -1,8 +1,10 @@ +from collections.abc import Iterator from pathlib import Path import pytest import strix.skills as skills_mod +from strix.agents.prompt import render_system_prompt from strix.skills import ( get_all_skill_names, get_available_skills, @@ -12,10 +14,11 @@ from strix.skills import ( skill_search_dirs, validate_requested_skills, ) +from strix.utils.resource_paths import get_strix_resource_path @pytest.fixture(autouse=True) -def _clear_extra_dirs() -> None: +def _clear_extra_dirs() -> Iterator[None]: original = list(skills_mod._EXTRA_SKILL_DIRS) skills_mod._EXTRA_SKILL_DIRS.clear() try: @@ -37,9 +40,11 @@ def _write_root_skill(root: Path, name: str, body: str) -> None: def test_no_registration_leaves_builtin_only() -> None: assert registered_skill_dirs() == () - builtin = skills_mod.get_strix_resource_path("skills") + builtin = get_strix_resource_path("skills") assert skill_search_dirs() == (builtin,) - assert {"nmap", "subfinder"}.issubset(get_available_skills()["tooling"]) + assert {"nmap", "subfinder"}.issubset( + {skill["name"] for skill in get_available_skills()["tooling"]} + ) def test_register_is_idempotent_and_ordered(tmp_path: Path) -> None: @@ -61,16 +66,125 @@ def test_registered_dir_adds_new_skill(tmp_path: Path) -> None: register_skill_dir(tmp_path) assert "widget" in get_all_skill_names() - assert get_available_skills()["extra"] == ["widget"] + assert get_available_skills()["extra"] == [{"name": "widget", "description": ""}] assert load_skills(["widget"]) == {"widget": "widget body"} +def test_available_skill_includes_frontmatter_description(tmp_path: Path) -> None: + _write_skill( + tmp_path, + "extra", + "widget", + "---\nname: widget\ndescription: Useful widget guidance\n---\nwidget body", + ) + register_skill_dir(tmp_path) + + assert get_available_skills()["extra"] == [ + {"name": "widget", "description": "Useful widget guidance"} + ] + + +def test_available_skill_supports_colon_in_description(tmp_path: Path) -> None: + _write_skill( + tmp_path, + "extra", + "widget", + '---\nname: widget\ndescription: "Useful widget: handles YAML"\n---\nwidget body', + ) + register_skill_dir(tmp_path) + + assert get_available_skills()["extra"] == [ + {"name": "widget", "description": "Useful widget: handles YAML"} + ] + + +def test_available_skill_normalizes_quoted_description(tmp_path: Path) -> None: + _write_skill( + tmp_path, + "extra", + "widget", + '---\nname: widget\ndescription: "Useful: widget guidance"\n---\nwidget body', + ) + register_skill_dir(tmp_path) + + assert get_available_skills()["extra"] == [ + {"name": "widget", "description": "Useful: widget guidance"} + ] + + +def test_available_skill_normalizes_multiline_descriptions(tmp_path: Path) -> None: + _write_skill( + tmp_path, + "extra", + "block", + "---\nname: block\n\ndescription: |\n" + " First paragraph\n\n Second paragraph\n\n---\nblock body", + ) + _write_skill( + tmp_path, + "extra", + "plain", + "---\nname: plain\n\ndescription: First line\n Second line\n\n---\nplain body", + ) + register_skill_dir(tmp_path) + + available = {skill["name"]: skill["description"] for skill in get_available_skills()["extra"]} + assert available == { + "block": "First paragraph Second paragraph", + "plain": "First line Second line", + } + + +def test_available_skill_supports_block_scalar_trailing_comment(tmp_path: Path) -> None: + _write_skill( + tmp_path, + "extra", + "commented", + "---\nname: commented\ndescription: | # paragraph\n" + " First line\n Second line\n---\ncommented body", + ) + register_skill_dir(tmp_path) + + assert get_available_skills()["extra"] == [ + {"name": "commented", "description": "First line Second line"} + ] + + +def test_malformed_frontmatter_keeps_skill_body(tmp_path: Path) -> None: + _write_skill( + tmp_path, + "extra", + "broken", + "---\nname: [broken\ndescription: should be empty\n---\nbroken body", + ) + register_skill_dir(tmp_path) + + assert get_available_skills()["extra"] == [{"name": "broken", "description": ""}] + assert load_skills(["extra/broken"]) == {"broken": "broken body"} + + +def test_system_prompt_renders_skill_descriptions() -> None: + prompt = render_system_prompt(scan_mode="quick", is_root=True) + + assert "- technologies/firebase: Firebase security testing covering" in prompt + + +def test_system_prompt_omits_empty_skill_description(tmp_path: Path) -> None: + _write_skill(tmp_path, "extra", "widget", "---\nname: widget\ndescription:\n---\nwidget body") + register_skill_dir(tmp_path) + + prompt = render_system_prompt(scan_mode="quick", is_root=True) + + assert "- extra/widget\n" in prompt + assert "- extra/widget: " not in prompt + + def test_registered_root_skill_is_discoverable_and_valid(tmp_path: Path) -> None: _write_root_skill(tmp_path, "widget", "widget body") register_skill_dir(tmp_path) assert "widget" in get_all_skill_names() - assert get_available_skills()["root"] == ["widget"] + assert get_available_skills()["root"] == [{"name": "widget", "description": ""}] assert validate_requested_skills(["widget"]) is None assert validate_requested_skills(["root/widget"]) is None assert load_skills(["widget"]) == {"widget": "widget body"} @@ -83,8 +197,8 @@ def test_ambiguous_bare_skill_requires_qualified_name(tmp_path: Path) -> None: register_skill_dir(tmp_path) assert "widget" in get_all_skill_names() - assert get_available_skills()["alpha"] == ["widget"] - assert get_available_skills()["beta"] == ["widget"] + assert get_available_skills()["alpha"] == [{"name": "widget", "description": ""}] + assert get_available_skills()["beta"] == [{"name": "widget", "description": ""}] assert validate_requested_skills(["alpha/widget"]) is None assert validate_requested_skills(["beta/widget"]) is None