Compare commits

...
6 changed files with 264 additions and 33 deletions
+4 -4
View File
@@ -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
+4 -2
View File
@@ -490,8 +490,10 @@ Default user: pentester (sudo available)
<available_skills>
On-demand specialist skills. Spawn a specialist via `create_agent(skills=[...])`, or pull guidance inline for yourself via `load_skill(skills=[...])`. Anything wrapped in `<specialized_knowledge>` 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 -%}
</available_skills>
{% endif %}
+71 -8
View File
@@ -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<body>.*?)\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 "<content>", error)
parsed = None
if not _is_frontmatter_mapping(parsed):
logger.warning("Skill frontmatter is not a mapping: %s", source or "<content>")
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)
+1 -1
View File
@@ -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**
@@ -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/<project>/databases/(default)/documents/<path>`
- Realtime DB: `https://<project>.firebaseio.com/.json`
- Storage REST: `https://storage.googleapis.com/storage/v1/b/<bucket>`
- GCS JSON API: `https://storage.googleapis.com/storage/v1/b/<bucket>`
- Firebase Storage rules API: `https://firebasestorage.googleapis.com/v0/b/<bucket>/o`
Cloud Storage has two front doors with different authorization engines:
| Front door | Authorization engine |
| --- | --- |
| `storage.googleapis.com/<bucket>/<object>` and `/storage/v1/b/<bucket>` | GCS IAM and per-object ACLs |
| `firebasestorage.googleapis.com/v0/b/<bucket>/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/<project>`)
@@ -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/<bucket>/o?prefix=<known-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 `<project>.appspot.com` and `<project>.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/<orgId>/...`). 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/<orgId>/...`). 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
+121 -7
View File
@@ -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