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. Third-party service and platform security.
| Skill | Coverage | | Skill | Coverage |
| -------------------- | ---------------------------------- | | ---------- | ------------------------------------------------------ |
| `supabase` | Supabase RLS bypasses, auth issues | | `supabase` | Supabase RLS bypasses, auth issues |
| `firebase_firestore` | Firestore rules, Firebase auth | | `firebase` | Firebase Firestore, Storage rules, Auth, and Functions |
### Protocols ### Protocols
+4 -2
View File
@@ -490,8 +490,10 @@ Default user: pentester (sudo available)
<available_skills> <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. 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 -%} {% for category, skills in available_skills | dictsort -%}
- {{ category }}: {{ names | join(', ') }} {% for skill in skills -%}
- {{ category }}/{{ skill.name }}{% if skill.description %}: {{ skill.description }}{% endif %}
{% endfor -%}
{% endfor -%} {% endfor -%}
</available_skills> </available_skills>
{% endif %} {% endif %}
+71 -8
View File
@@ -4,6 +4,9 @@ import threading
from collections import Counter from collections import Counter
from collections.abc import Iterator from collections.abc import Iterator
from pathlib import Path from pathlib import Path
from typing import TypeGuard
import yaml
from strix.telemetry import posthog, scarf from strix.telemetry import posthog, scarf
from strix.utils.resource_paths import get_strix_resource_path 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__) 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"}) _INTERNAL_SKILL_CATEGORIES: frozenset[str] = frozenset({"scan_modes", "coordination"})
_ROOT_SKILL_CATEGORY = "root" _ROOT_SKILL_CATEGORY = "root"
_EXTRA_SKILL_DIRS: list[Path] = [] _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: 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} 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("/") category, _, name = skill_name.partition("/")
for skills_dir in skill_search_dirs(): for skills_dir in skill_search_dirs():
candidate = _qualified_skill_file(skills_dir, category, name) candidate = _qualified_skill_file(skills_dir, category, name)
if candidate is not None: if candidate is not None:
return [candidate] return candidate
return [] 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]: def _bare_skill_files(skill_name: str) -> list[Path]:
@@ -145,10 +158,59 @@ def _bare_skill_files(skill_name: str) -> list[Path]:
return candidates return candidates
def get_available_skills() -> dict[str, list[str]]: def _parse_skill_content(content: str, source: Path | None = None) -> tuple[dict[str, str], str]:
grouped: dict[str, list[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(): 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 return grouped
@@ -228,7 +290,8 @@ def load_skills(skill_names: list[str]) -> dict[str, str]:
continue continue
var_name = skill_name.split("/")[-1] 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) logger.debug("Loaded skill: %s -> %s", skill_name, var_name)
_track_skill_loaded(var_name, file_path) _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** **Storage & Data**
- Cloud Storage (GCS) buckets and objects - 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 - Secret Manager, Cloud KMS keys
**Compute** **Compute**
@@ -1,9 +1,9 @@
--- ---
name: firebase-firestore name: firebase
description: Firebase/Firestore security testing covering security rules, Cloud Functions, and client-side trust issues 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. 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** **Endpoints**
- Firestore REST: `https://firestore.googleapis.com/v1/projects/<project>/databases/(default)/documents/<path>` - Firestore REST: `https://firestore.googleapis.com/v1/projects/<project>/databases/(default)/documents/<path>`
- Realtime DB: `https://<project>.firebaseio.com/.json` - 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** **Auth**
- Google-signed ID tokens (iss: `accounts.google.com` or `securetoken.google.com/<project>`) - 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 - Public reads on sensitive buckets/paths
- Signed URLs with long TTL, no content-disposition controls, replayable across tenants - Signed URLs with long TTL, no content-disposition controls, replayable across tenants
- List operations exposed: `/o?prefix=` enumerates object keys - 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** **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 - 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 - 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 ## Testing Methodology
1. **Extract config** - Get project config from client bundle 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, admin 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 3. **Build matrix** - Resource × Action × Principal across Firestore/Realtime/Storage/Functions
4. **SDK vs REST** - Exercise every action via both to detect parity gaps 4. **Exercise both Storage doors** - Test Firebase Storage rules endpoints separately from GCS IAM/ACL URLs
5. **Seed IDs** - Start from list/query paths to gather document IDs 5. **SDK vs REST** - Exercise every action via both to detect parity gaps
6. **Cross-principal** - Swap document paths, tenants, and user IDs across principals 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 ## Tooling
@@ -206,6 +257,7 @@ Apps often implement multi-tenant data models (`orgs/<orgId>/...`). Bind tenant
## Validation Requirements ## Validation Requirements
- Owner vs non-owner Firestore queries showing unauthorized access or metadata leak - 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` - Function accepting forged/foreign identity (wrong `aud`/`iss`) or trusting client `uid`/`orgId`
- Minimal reproducible requests with roles/tokens used and observed deltas - 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 from pathlib import Path
import pytest import pytest
import strix.skills as skills_mod import strix.skills as skills_mod
from strix.agents.prompt import render_system_prompt
from strix.skills import ( from strix.skills import (
get_all_skill_names, get_all_skill_names,
get_available_skills, get_available_skills,
@@ -12,10 +14,11 @@ from strix.skills import (
skill_search_dirs, skill_search_dirs,
validate_requested_skills, validate_requested_skills,
) )
from strix.utils.resource_paths import get_strix_resource_path
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def _clear_extra_dirs() -> None: def _clear_extra_dirs() -> Iterator[None]:
original = list(skills_mod._EXTRA_SKILL_DIRS) original = list(skills_mod._EXTRA_SKILL_DIRS)
skills_mod._EXTRA_SKILL_DIRS.clear() skills_mod._EXTRA_SKILL_DIRS.clear()
try: try:
@@ -37,9 +40,11 @@ def _write_root_skill(root: Path, name: str, body: str) -> None:
def test_no_registration_leaves_builtin_only() -> None: def test_no_registration_leaves_builtin_only() -> None:
assert registered_skill_dirs() == () assert registered_skill_dirs() == ()
builtin = skills_mod.get_strix_resource_path("skills") builtin = get_strix_resource_path("skills")
assert skill_search_dirs() == (builtin,) 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: 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) register_skill_dir(tmp_path)
assert "widget" in get_all_skill_names() 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"} 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: def test_registered_root_skill_is_discoverable_and_valid(tmp_path: Path) -> None:
_write_root_skill(tmp_path, "widget", "widget body") _write_root_skill(tmp_path, "widget", "widget body")
register_skill_dir(tmp_path) register_skill_dir(tmp_path)
assert "widget" in get_all_skill_names() 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(["widget"]) is None
assert validate_requested_skills(["root/widget"]) is None assert validate_requested_skills(["root/widget"]) is None
assert load_skills(["widget"]) == {"widget": "widget body"} 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) register_skill_dir(tmp_path)
assert "widget" in get_all_skill_names() assert "widget" in get_all_skill_names()
assert get_available_skills()["alpha"] == ["widget"] assert get_available_skills()["alpha"] == [{"name": "widget", "description": ""}]
assert get_available_skills()["beta"] == ["widget"] assert get_available_skills()["beta"] == [{"name": "widget", "description": ""}]
assert validate_requested_skills(["alpha/widget"]) is None assert validate_requested_skills(["alpha/widget"]) is None
assert validate_requested_skills(["beta/widget"]) is None assert validate_requested_skills(["beta/widget"]) is None