feat(stairwell): add Stairwell external-import connector (#7096) - #7098
Open
zachdaniel-SW wants to merge 2 commits into
Open
feat(stairwell): add Stairwell external-import connector (#7096)#7098zachdaniel-SW wants to merge 2 commits into
zachdaniel-SW wants to merge 2 commits into
Conversation
Contributor License AgreementHey @zachdaniel-SW! Thank you for your contribution to Filigran! Before we can merge this pull request, we need you to sign our Contributor License Agreement (CLA). Why do we need a CLA?The CLA helps protect both you and Filigran. It ensures that:
How to signYou can sign the CLA using either of these methods:
Once signed, this comment will be automatically updated. ❌ CLA not signed yet This is an automated message from the Filigran CLA Bot. If you have questions, please contact the maintainers. |
Author
|
Video of integration - https://fathom.video/share/GYDfsKAdPCe722ar7Uvyqzo6dAP-EZMt |
Contributor
There was a problem hiding this comment.
Pull request overview
Adds a new external-import/stairwell connector that periodically imports Stairwell MalEval-derived indicators (file hashes plus extracted IP/domain/URL observables) into OpenCTI using deterministic STIX IDs, with per-run wrapper objects (Grouping/Report), stateful cutoffs, and unit tests.
Changes:
- Implemented Stairwell API client, CEL filtering + pagination, and STIX bundle construction for Indicators/SCOs/relationships.
- Added connector configuration (Pydantic/connectors-sdk settings), runnable entrypoint, and Docker/deploy assets.
- Added pytest coverage for filter/cutoff logic and the import runner’s bundle emission behavior.
Reviewed changes
Copilot reviewed 22 out of 24 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| external-import/stairwell/src/tests/test-requirements.txt | Defines test dependencies for the Stairwell connector test suite. |
| external-import/stairwell/src/tests/test_import_runner.py | Unit tests validating end-to-end runner behavior and emitted STIX shapes. |
| external-import/stairwell/src/tests/test_import_filter.py | Unit tests for duration parsing, cutoff computation, and CEL filter shape. |
| external-import/stairwell/src/tests/conftest.py | Ensures src/ is importable during tests. |
| external-import/stairwell/src/requirements.txt | Runtime dependencies for the connector. |
| external-import/stairwell/src/main.py | Thin runtime entrypoint wiring settings → helper → connector run loop. |
| external-import/stairwell/src/connector/stix_builder.py | Deterministic ID generation and helpers for building STIX objects/bundles. |
| external-import/stairwell/src/connector/stairwell/score.py | Maps Stairwell probability buckets to confidence scores. |
| external-import/stairwell/src/connector/stairwell/client.py | HTTP client for Stairwell API endpoints with retries and helpers. |
| external-import/stairwell/src/connector/stairwell/init.py | Exposes Stairwell client/score symbols for internal imports. |
| external-import/stairwell/src/connector/settings.py | Connectors-sdk/Pydantic settings model for Stairwell + external-import config. |
| external-import/stairwell/src/connector/import_runner.py | Core import loop: pagination, filtering, STIX construction, state updates, emission. |
| external-import/stairwell/src/connector/import_filter.py | Builds CEL filter and computes cutoffs from connector state/first-run window. |
| external-import/stairwell/src/connector/connector.py | Connector class integrating scheduler + work item registration with the runner. |
| external-import/stairwell/src/connector/init.py | Connector package marker. |
| external-import/stairwell/src/config.yml.sample | Sample YAML configuration for local deployments. |
| external-import/stairwell/README.md | Connector documentation (setup/config/behavior/tests). |
| external-import/stairwell/entrypoint.sh | Container entrypoint script. |
| external-import/stairwell/Dockerfile | Docker build for the connector image. |
| external-import/stairwell/docker-compose.yml | Example compose service definition for deployment. |
| external-import/stairwell/.env.test.sample | Sample env exports for local manual testing. |
| external-import/stairwell/.dockerignore | Excludes tests/config from Docker build context. |
| external-import/stairwell/metadata/connector_manifest.json | Connector manifest metadata for OpenCTI connectors catalog. |
Comments suppressed due to low confidence (4)
external-import/stairwell/src/connector/import_runner.py:112
- indicator_ids.append(...) and run_finished are both unused. Removing them reduces noise and avoids unused-variable warnings.
indicator_ids.append(new_indicator_id)
total_emitted += 1
run_finished = datetime.now(tz=timezone.utc)
date_label = run_started.strftime("%Y-%m-%d")
external-import/stairwell/README.md:60
- This says min-bucket filtering is pushed to Stairwell via CEL, but the implementation explicitly avoids server-side bucket clauses (client-side filtering only). The README should match actual behavior to avoid misleading operators.
- **CEL filtering.** Min-bucket and time cutoff are pushed to Stairwell as a
CEL expression so the API only returns relevant rows.
external-import/stairwell/README.md:65
- The code sets valid_from from Stairwell first-seen/upload timestamps (falling back to run start), not always the ingest time as documented here.
- **Indicator validity.** Each Indicator gets `valid_from` = ingest time,
`valid_until` = `valid_from + import_indicator_validity_days`.
external-import/stairwell/src/tests/test-requirements.txt:2
- CI's run_test.sh installs only the requirements from this file before running pytest. To ensure tests run in an environment that matches the connector runtime (and catch missing deps), include the connector's src/requirements.txt here (common pattern in this repo).
pytest~=8.3
Comment on lines
+197
to
+201
| if status >= 400 or not isinstance(data, dict): | ||
| self.helper.log_warning( | ||
| f"Stairwell list_objects_metadata returned status {status} on page {page_num}; stopping." | ||
| ) | ||
| return |
Comment on lines
+349
to
+353
| make_external_reference( | ||
| "Stairwell", | ||
| f"{self.client._base_url}/search?search-query={value}", | ||
| f"Stairwell intel for {value}", | ||
| ) |
Comment on lines
+356
to
+360
| pattern=f"[{obs_path} = '{value}']", | ||
| name=f"Stairwell {kind}: {value}", | ||
| seed=f"stairwell-{kind}-indicator|{value.lower()}", | ||
| valid_from=valid_from_iso, | ||
| valid_until=valid_until_iso, |
Comment on lines
+74
to
+76
| objects: list[dict[str, Any]] = [stairwell_identity()] | ||
| indicator_ids: list[str] = [] | ||
| sco_ids_seen: set[str] = set() |
Comment on lines
+58
to
+62
| def build_cel_filter( | ||
| cutoff: datetime | None, | ||
| min_bucket: str | None = None, | ||
| scope_environment: bool = False, | ||
| ) -> str: |
Author
There was a problem hiding this comment.
Good catch — import_scope isn't wired into the CEL filter yet. The Stairwell API doesn't expose a per-environment CEL identifier we can rely on (server-side bucket/'in' queries also return 500), so we've kept the parameter for forward-compatibility and updated the README, config samples, setting description, and PR body to state it's reserved and not yet effective (every run currently queries the full corpus). We'll implement the scoping clause once the correct field is available.
| | Max indicators | `stairwell.import_max_indicators` | `STAIRWELL_IMPORT_MAX_INDICATORS` | no | `1000` | Cap per run | | ||
| | Page size | `stairwell.import_page_size` | `STAIRWELL_IMPORT_PAGE_SIZE` | no | `100` | API page size | | ||
| | Indicator validity | `stairwell.import_indicator_validity_days` | `STAIRWELL_IMPORT_INDICATOR_VALIDITY_DAYS` | no | `90` | `valid_until` offset | | ||
| | Min bucket | `stairwell.import_min_bucket` | `STAIRWELL_IMPORT_MIN_BUCKET` | no | `HIGH` | `LOW`, `MEDIUM`, `HIGH`, `MALICIOUS` | |
Comment on lines
+70
to
+73
| import_min_bucket: str = Field( | ||
| description="Minimum MalEval bucket: LOW | MEDIUM | HIGH | MALICIOUS.", | ||
| default="HIGH", | ||
| ) |
| import_max_indicators: 1000 # optional (default: 1000) | ||
| import_page_size: 100 # optional (default: 100) | ||
| import_indicator_validity_days: 90 # optional (default: 90) | ||
| import_min_bucket: 'HIGH' # optional, one of: LOW, MEDIUM, HIGH, MALICIOUS (default: 'HIGH') |
| export STAIRWELL_IMPORT_MAX_INDICATORS=5 # cap for manual testing; raise for real use | ||
| export STAIRWELL_IMPORT_PAGE_SIZE=20 | ||
| export STAIRWELL_IMPORT_INDICATOR_VALIDITY_DAYS=90 | ||
| export STAIRWELL_IMPORT_MIN_BUCKET="HIGH" # LOW | MEDIUM | HIGH | MALICIOUS |
…ator ids, bucket/scope/validity doc accuracy
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Proposed changes
Adds
external-import/stairwell, an EXTERNAL_IMPORT connector that periodicallyimports high-confidence Stairwell MalEval indicators (file/IP/domain/URL) into
OpenCTI as Indicators + related observables, wrapped in a per-run Grouping or
Report. Configurable probability bucket and time window. (An environment/global
scope parameter is accepted for forward-compatibility but is not yet effective —
every run currently queries the full corpus.)
Built to the current template: connectors-sdk + Pydantic settings, manifest,
entrypoint, thin main.py. Scheduling via helper.schedule_process with per-run
work registration (initiate_work / to_processed). STIX emitted via bundles with
deterministic IDs and cleanup_inconsistent_bundle=True.
Demo
Workflow walkthrough (both Stairwell connectors):
https://fathom.video/share/GYDfsKAdPCe722ar7Uvyqzo6dAP-EZMt
Related issues
Close #7096
Checklist
Further comments
Unit tests: 21 passing (
pytest tests/). Config layer verified against realpycti 7.260722.0 + connectors-sdk. Live end-to-end validation on OpenCTI 6.8.x
in progress.