mirror of
https://github.com/reconurge/flowsint.git
synced 2026-07-11 09:12:32 -05:00
chore: update .gitignore
This commit is contained in:
dextmorgn
178
.claude/skills/flowsint-enricher-builder/SKILL.md
Normal file
178
.claude/skills/flowsint-enricher-builder/SKILL.md
Normal file
@@ -0,0 +1,178 @@
|
||||
---
|
||||
name: flowsint-enricher-builder
|
||||
description: Expert guidance for building Flowsint enrichers and their supporting types. Use when the user wants to add a new enricher, create a new Flowsint type, wire a new external API/tool into Flowsint, debug type/enricher discovery, or design a pivot from entity A to entity B. Knows where types live, how the enricher base class works, how vault secrets and params resolve, and when to recommend creating a new type instead of forcing data into an existing one.
|
||||
---
|
||||
|
||||
# Flowsint Enricher Builder
|
||||
|
||||
You build enrichers and types for Flowsint. You do not memorize the catalog — you know where to look and how the pieces fit. Always read source before generating code: type definitions and existing enrichers are the ground truth.
|
||||
|
||||
## Authoritative source paths
|
||||
|
||||
Read these first. Never assume signatures or fields — open the file.
|
||||
|
||||
| What | Path |
|
||||
|---|---|
|
||||
| Type definitions | `flowsint-types/src/flowsint_types/<name>.py` |
|
||||
| Type registry + decorator | `flowsint-types/src/flowsint_types/registry.py` |
|
||||
| Type package exports | `flowsint-types/src/flowsint_types/__init__.py` |
|
||||
| Enricher base class | `flowsint-core/src/flowsint_core/core/enricher_base.py` |
|
||||
| Enricher registry + decorator | `flowsint-enrichers/src/flowsint_enrichers/registry.py` |
|
||||
| Existing enrichers (templates) | `flowsint-enrichers/src/flowsint_enrichers/<input_type>/to_<output>.py` |
|
||||
| UI category mapping | `flowsint-core/src/flowsint_core/core/services/type_registry_service.py` (`_get_category_definitions`) |
|
||||
| Vault interface | `flowsint-core/src/flowsint_core/core/vault.py` |
|
||||
| Logger interface | `flowsint-core/src/flowsint_core/core/logger.py` |
|
||||
| Tools (external CLI/API wrappers) | `tools/` (top-level), e.g. `tools.network.subfinder.SubfinderTool` |
|
||||
| Doc — types tutorial | `docs/developers/managing-types.mdx` |
|
||||
| Doc — enrichers tutorial | `docs/developers/managing-enrichers.mdx` |
|
||||
| Doc — enricher catalog | `docs/sources/available-enrichers.mdx` |
|
||||
|
||||
## The first question: new type or reuse?
|
||||
|
||||
When the user describes a enricher, decide before writing code:
|
||||
|
||||
1. **List the entities involved** — input data, output data, intermediate fields you'll attach.
|
||||
2. **For each, check `flowsint-types/src/flowsint_types/`** — open the closest candidate file and read its fields.
|
||||
3. **Decide:**
|
||||
- **Reuse** if existing type covers all required fields (extras allowed — `ConfigDict.extra = "allow"`).
|
||||
- **Extend an existing type** if 1–2 fields are missing — propose adding optional fields to the existing model.
|
||||
- **Create new type** if the entity is conceptually distinct (different primary key, different label semantics, different graph role).
|
||||
4. **Never cram data into a wrong type.** If a "Domain" enricher returns risk scores, a `RiskProfile` exists — don't stuff scores into `Domain` metadata. If nothing fits, propose a new type and tell the user why.
|
||||
|
||||
Surface the decision to the user before generating code: list candidate types you found, what's missing, and your recommendation.
|
||||
|
||||
## Anatomy of an enricher
|
||||
|
||||
Minimum surface (read `enricher_base.py` for the full contract):
|
||||
|
||||
```python
|
||||
from typing import List
|
||||
from flowsint_core.core.enricher_base import Enricher
|
||||
from flowsint_core.core.logger import Logger
|
||||
from flowsint_enrichers.registry import flowsint_enricher
|
||||
from flowsint_types import Domain, Ip # or whatever types
|
||||
|
||||
@flowsint_enricher
|
||||
class MyEnricher(Enricher):
|
||||
"""[Source name] One-line purpose."""
|
||||
|
||||
InputType = Domain # base type, not List[Domain]
|
||||
OutputType = Ip
|
||||
|
||||
@classmethod
|
||||
def name(cls) -> str: return "domain_to_ip" # snake_case, unique
|
||||
@classmethod
|
||||
def category(cls) -> str: return "Domain" # see note on casing below
|
||||
@classmethod
|
||||
def key(cls) -> str: return "domain" # primary field of InputType
|
||||
|
||||
@classmethod
|
||||
def get_params_schema(cls): # optional, only if params needed
|
||||
return [...]
|
||||
|
||||
async def scan(self, data: List[InputType]) -> List[OutputType]:
|
||||
...
|
||||
|
||||
def postprocess(self, results, input_data):
|
||||
for src, dst in zip(input_data, results):
|
||||
self.create_node(src)
|
||||
self.create_node(dst)
|
||||
self.create_relationship(src, dst, "RESOLVES_TO")
|
||||
return results
|
||||
|
||||
InputType = MyEnricher.InputType
|
||||
OutputType = MyEnricher.OutputType
|
||||
```
|
||||
|
||||
**File location:** `flowsint-enrichers/src/flowsint_enrichers/<input_type>/to_<target>.py`. The directory matches the input type's lowercase name. If no directory exists for your input type, create it (no `__init__.py` needed — auto-discovery walks the tree).
|
||||
|
||||
**Registration:** the `@flowsint_enricher` decorator does it. Do not edit any `registry.py`. API restart picks up new files via `load_all_enrichers()`.
|
||||
|
||||
## Params and secrets
|
||||
|
||||
Defined via `get_params_schema()` classmethod. Each entry is a dict:
|
||||
|
||||
| Field | Required | Notes |
|
||||
|---|---|---|
|
||||
| `name` | yes | Param key; for `vaultSecret`, also the default vault key name |
|
||||
| `type` | yes | One of `string`, `number`, `select`, `url`, `vaultSecret` |
|
||||
| `description` | yes | Shown in UI |
|
||||
| `required` | no | Defaults to `false` |
|
||||
| `default` | no | Default value |
|
||||
| `options` | for `select` | List of `{"label": ..., "value": ...}` |
|
||||
|
||||
Read params inside `scan()`:
|
||||
|
||||
```python
|
||||
mode = self.params.get("mode", "passive")
|
||||
api_key = self.get_secret("MY_API_KEY") # vault-resolved during async_init
|
||||
```
|
||||
|
||||
**Vault resolution flow** (see `Enricher.resolve_params` in `enricher_base.py`):
|
||||
1. If user passed a vault ID in params → vault looked up by that ID.
|
||||
2. Else → vault looked up by the param name (e.g. `MY_API_KEY`).
|
||||
3. If `required: true` and nothing found → `Exception("Required vault secret 'MY_API_KEY' is missing...")`.
|
||||
|
||||
**Never hardcode keys.** Always declare a `vaultSecret` param. Document the expected vault key name in the docstring.
|
||||
|
||||
## Graph operations (postprocess)
|
||||
|
||||
`create_node(obj)` and `create_relationship(from_obj, to_obj, rel_label="IS_RELATED_TO")` take Pydantic objects directly. Don't manually construct node dicts — pass the typed instance.
|
||||
|
||||
Relationship label convention: `UPPER_SNAKE_CASE` verb phrase (`HAS_DOMAIN`, `RESOLVES_TO`, `FOUND_IN_BREACH`). Be consistent with existing enrichers — grep before inventing a new label.
|
||||
|
||||
`self.log_graph_message("...")` for graph-related progress logs. `Logger.info / error / warn(self.sketch_id, {"message": "..."})` for general logs.
|
||||
|
||||
## Creating a new type — checklist
|
||||
|
||||
When you decide a new type is warranted:
|
||||
|
||||
1. **File**: `flowsint-types/src/flowsint_types/<snake_case>.py`.
|
||||
2. **Class**: `PascalCase`, inherit from `FlowsintType`, decorate with `@flowsint_type`.
|
||||
3. **Exactly one primary field**: `Field(..., json_schema_extra={"primary": True})`. Must uniquely identify the entity (used as Neo4j MERGE key).
|
||||
4. **`compute_label`**: `@model_validator(mode='after')`, sets `self.nodeLabel`, returns `self`. Handle `None` for optional fields.
|
||||
5. **Export in `__init__.py`**: add import + entry in `__all__`.
|
||||
6. **Category** (optional but recommended): add a `("MyType", "primary_field_name", icon)` tuple in `_get_category_definitions()` in `type_registry_service.py`. Without this, the type works as an enricher I/O but doesn't show in the UI type picker.
|
||||
7. **Reinstall**: `cd flowsint-types && poetry install` (or `make prod` from repo root).
|
||||
8. **Test**: write a `tests/test_<name>.py` covering creation, primary uniqueness, `compute_label` with full/partial fields.
|
||||
|
||||
Full template + patterns: `docs/developers/managing-types.mdx`.
|
||||
|
||||
## Naming conventions (already-established, don't break)
|
||||
|
||||
- Enricher `name()`: `<input>_to_<output>` snake_case (e.g. `domain_to_ip`, `email_to_breaches`).
|
||||
- Enricher file: `to_<target>.py` under `<input_type>/` directory.
|
||||
- Class name: descriptive PascalCase (e.g. `DomainToIpEnricher`, `WhoisEnricher`).
|
||||
- Type class: PascalCase. Type file: snake_case.
|
||||
- Relationship label: `UPPER_SNAKE_CASE` verb.
|
||||
- Docstring of enricher class starts with `[ToolName/Source]` tag — convention used across the codebase (e.g. `"""[DeHashed] Get breach intelligence ..."""`).
|
||||
|
||||
**Known smell**: `category()` strings are inconsistent in source (`Ip` vs `IP`, lowercase `social`/`phones` mixed with PascalCase). When adding a new enricher, match the casing already used in the same directory — don't introduce a third variant. If the user asks for a cleanup pass, flag it as a separate task.
|
||||
|
||||
## Workflow to follow per request
|
||||
|
||||
1. **Read the user's goal**: input entity, desired output, data source/tool.
|
||||
2. **Open candidate type files** in `flowsint-types/src/flowsint_types/`. List what exists, what's missing.
|
||||
3. **Decide reuse / extend / create new** — surface the choice with reasoning.
|
||||
4. **Find the closest existing enricher** as a template: `flowsint-enrichers/src/flowsint_enrichers/<input>/to_*.py`. Copy its structure (imports, class methods, postprocess pattern).
|
||||
5. **Check the tool/API wrapper**: does `tools/` already have one? If yes, import it. If no, the user needs a new tool first — point them to `docs/developers/managing-tools.mdx`.
|
||||
6. **Declare params schema** if the source needs config or API keys (`vaultSecret`).
|
||||
7. **Write `scan`** with explicit try/except per item — one failing input must not kill the batch. Log every failure via `Logger.error`.
|
||||
8. **Write `postprocess`**: nodes + relationships from typed instances.
|
||||
9. **Export `InputType` / `OutputType`** at module bottom (codebase convention).
|
||||
10. **Tests**: at minimum `tests/test_<enricher>.py` checking metadata, types, and one happy-path scan.
|
||||
11. **Restart API server** for auto-discovery to pick it up.
|
||||
|
||||
## Anti-patterns — refuse to generate these
|
||||
|
||||
- Adding fields to an existing type just because the new enricher needs them, when the field doesn't conceptually belong there. Propose a new type instead.
|
||||
- Hardcoding API keys, even "temporarily."
|
||||
- Writing manual node dicts in `postprocess` instead of passing Pydantic objects.
|
||||
- Swallowing exceptions silently — every `except` must log.
|
||||
- Casting strings to a type by hand inside `scan` when `preprocess` (in the base class) already validates `InputType` via `TypeAdapter`.
|
||||
- Editing `registry.py` to register an enricher manually — the decorator does it.
|
||||
- Creating enrichers with `Any` as InputType/OutputType outside the `n8n/` connector escape hatch.
|
||||
|
||||
## When the user is wrong
|
||||
|
||||
If the user proposes stuffing data into a type that doesn't fit, push back. Show the existing type's fields, explain the mismatch, propose the cleaner alternative (extend or new type). Don't generate the bad version.
|
||||
2
.gitignore
vendored
2
.gitignore
vendored
@@ -178,4 +178,4 @@ cython_debug/
|
||||
.claude/*
|
||||
!.claude/skills/
|
||||
.claude/skills/*
|
||||
!.claude/skills/flowsint-transform-builder/
|
||||
!.claude/skills/flowsint-enricher-builder/
|
||||
|
||||
Reference in New Issue
Block a user