+1

docs/SUMMARY.md

View file Reviewed

··· 26 26 # Concepts & More 27 27 28 28 - [Writing Rules](rules.md) 29 29 + - [Integrations & Plugins](integrations.md) 29 30 - [User Research & Personas](research-personas.md) 30 31 31 32 ---

+312

docs/integrations.md

View file Reviewed

··· 1 1 + # Integrations & Plugins 2 2 + 3 3 + Osprey is designed to be extended without modifying the core codebase; you can wire up your own logic such as detection functions, output destinations, entity state storage, and ML models through plugin packages that Osprey discovers at startup. 4 4 + 5 5 + See the [`example_plugins/` directory](https://github.com/roostorg/osprey/tree/main/example_plugins) for reference. 6 6 + 7 7 + ## How plugins are loaded 8 8 + 9 9 + Osprey uses [pluggy](https://pluggy.readthedocs.io/) for plugin discovery. Your plugin package declares one or both of these entry-point groups in its `pyproject.toml`: 10 10 + 11 11 + - `osprey_plugin`: loaded by the standard gevent worker 12 12 + - `osprey_async_plugin`: loaded by the experimental asyncio worker 13 13 + 14 14 + For example: 15 15 + 16 16 + ```toml 17 17 + [project.entry-points.osprey_plugin] 18 18 + register_plugins = "register_plugins" 19 19 + 20 20 + [project.entry-points.osprey_async_plugin] 21 21 + register_async_plugins = "register_async_plugins" 22 22 + ``` 23 23 + 24 24 + Each entry point resolves to a module that contains hook functions decorated with `@hookimpl_osprey` or `@hookimpl_osprey_async`. Osprey calls each hook at startup to collect your registrations; see [`example_plugins/src/register_plugins.py`](https://github.com/roostorg/osprey/blob/main/example_plugins/src/register_plugins.py) and [`register_async_plugins.py`](https://github.com/roostorg/osprey/blob/main/example_plugins/src/register_async_plugins.py) for more. 25 25 + 26 26 + ## Writing UDFs 27 27 + 28 28 + A user-defined function (UDF) is a Python class that can be called from your rules. UDFs encapsulate reusable detection logic such as text matching, DNS lookups, hash comparisons, or ML inference and make it available under a named function in the rules language. 29 29 + 30 30 + ### Anatomy of a UDF 31 31 + 32 32 + UDFs require: 33 33 + 34 34 + 1. An **arguments** class (subclass of `ArgumentsBase`) that declares the parameters the UDF accepts with types 35 35 + 2. A **UDF class** (subclass of `UDFBase[Arguments, ReturnType]`) with an `execute` method that contains the logic 36 36 + 37 37 + For example: 38 38 + 39 39 + ```python 40 40 + # example_plugins/src/udfs/text_contains.py 41 41 + import re 42 42 + 43 43 + from osprey.engine.executor.execution_context import ExecutionContext 44 44 + from osprey.engine.udf.arguments import ArgumentsBase 45 45 + from osprey.engine.udf.base import UDFBase 46 46 + 47 47 + 48 48 + class TextContainsArguments(ArgumentsBase): 49 49 + text: str 50 50 + phrase: str 51 51 + case_sensitive = False 52 52 + 53 53 + 54 54 + class TextContains(UDFBase[TextContainsArguments, bool]): 55 55 + def execute(self, execution_context: ExecutionContext, arguments: TextContainsArguments) -> bool: 56 56 + escaped = re.escape(arguments.phrase) 57 57 + 58 58 + pattern = rf'\b{escaped}\b' 59 59 + 60 60 + flags = 0 if arguments.case_sensitive else re.IGNORECASE 61 61 + regex = re.compile(pattern, flags) 62 62 + 63 63 + return bool(regex.search(arguments.text)) 64 64 + ``` 65 65 + 66 66 + Once registered, `TextContains` is callable from rules as: 67 67 + 68 68 + ```python 69 69 + TextContains(text=SomeFeature, phrase="spam") 70 70 + ``` 71 71 + 72 72 + ### UDFs with side effects 73 73 + 74 74 + UDFs can also produce **effects**: structured outputs that downstream systems act on, such as banning a user or flagging content. Effects are expressed using `EffectBase` as the return type. See [`example_plugins/src/udfs/ban_user.py`](https://github.com/roostorg/osprey/blob/main/example_plugins/src/udfs/ban_user.py) for an example. 75 75 + 76 76 + ### Async UDFs 77 77 + 78 78 + UDFs that perform I/O (e.g. network calls or database reads) should subclass `AsyncUDFBase` when used in the async worker; see [`osprey_async_worker/src/osprey/async_worker/stdlib_udfs/async_mx_lookup.py`](https://github.com/roostorg/osprey/blob/main/osprey_async_worker/src/osprey/async_worker/stdlib_udfs/async_mx_lookup.py) for an example. Pure-computation UDFs like `TextContains` can be reused in both workers without modification. 79 79 + 80 80 + ### Registering UDFs 81 81 + 82 82 + Return your UDF classes from the `register_udfs` hook; for example: 83 83 + 84 84 + ```python 85 85 + from osprey.worker.adaptor.plugin_manager import hookimpl_osprey 86 86 + 87 87 + @hookimpl_osprey 88 88 + def register_udfs(): 89 89 + return [TextContains, BanUser] 90 90 + ``` 91 91 + 92 92 + ## Built-in UDFs: hash lookups 93 93 + 94 94 + Osprey's standard library includes hash UDFs (`HashMd5`, `HashSha1`, `HashSha256`, `HashSha512`) under the `HASH` category. They take a string `input` and return the hex digest. Use them in rules to compare hashed values against known-bad hash sets without storing raw data: 95 95 + 96 96 + ```python 97 97 + HashSha256(input=Username) == "e3b0c44298fc1c149afbf4c8996fb924..." 98 98 + ``` 99 99 + 100 100 + These are available without registration. 101 101 + 102 102 + ## Configuring input sinks 103 103 + 104 104 + An input sink is where events _enter_ Osprey. Osprey ships with built-in sources (Kafka, Google Pub/Sub, the Osprey Coordinator, and a synthetic generator for local testing) selected via the `InputStreamSource` config value. If none of those fit your platform, you can register a custom input stream as a plugin. 105 105 + 106 106 + ### Built-in sources 107 107 + 108 108 + The worker picks an input stream based on `InputStreamSource`: 109 109 + 110 110 + Source | Config | Use case 111 111 + -------------------- | ------------------------------------------------------------------- | ------------------------------------------------------------------------------ 112 112 + `KAFKA` | `OSPREY_KAFKA_INPUT_STREAM_TOPIC`, `OSPREY_KAFKA_BOOTSTRAP_SERVERS` | Consume Action events from a Kafka topic 113 113 + `PUBSUB` | `PUBSUB_OSPREY_PROJECT_ID`, `PUBSUB_OSPREY_RULES_SINK_SUBSCRIPTION` | Consume from Google Pub/Sub 114 114 + `OSPREY_COORDINATOR` | `OSPREY_COORDINATOR_SERVICE_NAME` | Pull work from the Osprey Coordinator service 115 115 + `SYNTHETIC` |   | Generates random fake events; useful for local dev without any upstream system 116 116 + `PLUGIN` |   | Delegates to your registered `register_input_stream` hook 117 117 + 118 118 + Set `InputStreamSource.KAFKA` (or whichever fits your existing infrastructure) if you already have events flowing through Kafka or Pub/Sub. Otherwise, implement a custom input stream and set `InputStreamSource.PLUGIN` in your config. 119 119 + 120 120 + ### Writing a custom input stream 121 121 + 122 122 + If your event source isn't Kafka or Pub/Sub (e.g. it's a webhook receiver, a different message queue, or a polling API), subclass `BaseInputStream` and implement `_gen`, a generator that yields one `Action` (wrapped in an `AckingContext`) per event. For example: 123 123 + 124 124 + ```python 125 125 + from collections.abc import Iterator 126 126 + 127 127 + from osprey.engine.executor.execution_context import Action 128 128 + from osprey.worker.sinks.sink.input_stream import BaseInputStream 129 129 + from osprey.worker.sinks.utils.acking_contexts import BaseAckingContext, NoopAckingContext 130 130 + 131 131 + 132 132 + class MyInputStream(BaseInputStream[BaseAckingContext[Action]]): 133 133 + def __init__(self, my_client): 134 134 + super().__init__() 135 135 + self._client = my_client 136 136 + 137 137 + def _gen(self) -> Iterator[BaseAckingContext[Action]]: 138 138 + while True: 139 139 + raw_event = self._client.poll() # block until the next event 140 140 + action = Action( 141 141 + action_id=int(raw_event['id']), 142 142 + action_name=raw_event['type'], 143 143 + data=raw_event['payload'], 144 144 + timestamp=raw_event['timestamp'], 145 145 + ) 146 146 + yield NoopAckingContext(item=action) 147 147 + ``` 148 148 + 149 149 + `_gen` is called once and re-used. It should block and yield indefinitely rather than returning. Use `NoopAckingContext` unless your source needs explicit ack/nack (e.g. a queue with at-least-once delivery), in which case implement a custom `BaseAckingContext` that acks on success. 150 150 + 151 151 + Register it from the hook, and set `InputStreamSource.PLUGIN` in your config so the worker picks it up; for example: 152 152 + 153 153 + ```python 154 154 + @hookimpl_osprey 155 155 + def register_input_stream(config): 156 156 + return MyInputStream(my_client=build_client(config)) 157 157 + ``` 158 158 + 159 159 + ## Configuring output sinks 160 160 + 161 161 + An output sink receives every `ExecutionResult` after rule evaluation and decides what to do with it, i.e. log it, forward it to a queue, call a webhook, or write to a database. 162 162 + 163 163 + ### Sync output sink 164 164 + 165 165 + Subclass `BaseOutputSink` and implement three methods; for example: 166 166 + 167 167 + ```python 168 168 + from osprey.worker.sinks.sink.output_sink import BaseOutputSink 169 169 + from osprey.engine.executor.execution_context import ExecutionResult 170 170 + 171 171 + 172 172 + class MyOutputSink(BaseOutputSink): 173 173 + def will_do_work(self, result: ExecutionResult) -> bool: 174 174 + # Return False to skip this result early (e.g. filter by rule hit) 175 175 + return True 176 176 + 177 177 + def push(self, result: ExecutionResult) -> None: 178 178 + # Do something with the result — send to a queue, call an API, etc. 179 179 + pass 180 180 + 181 181 + def stop(self) -> None: 182 182 + # Clean up connections, flush buffers 183 183 + pass 184 184 + ``` 185 185 + 186 186 + Register it from the hook; for example: 187 187 + 188 188 + ```python 189 189 + @hookimpl_osprey 190 190 + def register_output_sinks(config): 191 191 + return [MyOutputSink()] 192 192 + ``` 193 193 + 194 194 + ### Async output sink 195 195 + 196 196 + For the async worker, subclass `AsyncBaseOutputSink` and make `push` and `stop` coroutines. See [`example_plugins/src/async_sinks/example_async_output_sink.py`](https://github.com/roostorg/osprey/blob/main/example_plugins/src/async_sinks/example_async_output_sink.py); for example: 197 197 + 198 198 + ```python 199 199 + from osprey.async_worker.adaptor.interfaces import AsyncBaseOutputSink 200 200 + import logging 201 201 + 202 202 + logger = logging.getLogger(__name__) 203 203 + 204 204 + class ExampleAsyncOutputSink(AsyncBaseOutputSink): 205 205 + def will_do_work(self, result: ExecutionResult) -> bool: 206 206 + return True 207 207 + 208 208 + async def push(self, result: ExecutionResult) -> None: 209 209 + logger.info( 210 210 + 'example async output sink: features=%s verdicts=%s', 211 211 + result.extracted_features_json, 212 212 + result.verdicts, 213 213 + ) 214 214 + 215 215 + async def stop(self) -> None: 216 216 + pass 217 217 + ``` 218 218 + 219 219 + Register it with `@hookimpl_osprey_async` under the hook name `register_async_output_sinks`. This is a different hook from the sync `register_output_sinks` above, and goes in your `register_async_plugins.py` module (the one wired to the `osprey_async_plugin` entry point); for example: 220 220 + 221 221 + ```python 222 222 + from osprey.async_worker.adaptor.plugin_manager import hookimpl_osprey_async 223 223 + 224 224 + @hookimpl_osprey_async 225 225 + def register_async_output_sinks(config): 226 226 + return [ExampleAsyncOutputSink()] 227 227 + ``` 228 228 + 229 229 + ## Connecting to a review tool via a labels service 230 230 + 231 231 + Osprey tracks state across events through entity labels: arbitrary tags attached to users, accounts, or other entities. Labels are read during rule evaluation and written by rules with label effects. To persist labels across process restarts (and share them between workers), you provide a `LabelsServiceBase` implementation backed by your own storage. 232 232 + 233 233 + The example implementation in [`example_plugins/src/services/labels_service.py`](https://github.com/roostorg/osprey/blob/main/example_plugins/src/services/labels_service.py) uses PostgreSQL, e.g.: 234 234 + 235 235 + ```python 236 236 + from osprey.worker.lib.storage.labels import LabelsServiceBase 237 237 + 238 238 + class PostgresLabelsService(LabelsServiceBase): 239 239 + def initialize(self) -> None: 240 240 + # Called once at startup — open connections here 241 241 + ... 242 242 + 243 243 + def read_labels(self, entity) -> EntityLabels: 244 244 + # Return labels for this entity from your store 245 245 + ... 246 246 + 247 247 + @contextmanager 248 248 + def read_modify_write_labels_atomically(self, entity): 249 249 + # Yield the current labels; caller mutates them in place; 250 250 + # persist the result before the context manager exits 251 251 + ... 252 252 + ``` 253 253 + 254 254 + Register it from the hook: 255 255 + 256 256 + ```python 257 257 + @hookimpl_osprey 258 258 + def register_labels_service_or_provider(config): 259 259 + return PostgresLabelsService() 260 260 + ``` 261 261 + 262 262 + A labels service backed by your existing datastore lets Osprey decisions feed directly into your review tool: label an entity "flagged", and your review queue queries for that label. 263 263 + 264 264 + ## Plugging in your own ML model 265 265 + 266 266 + ML models integrate as UDFs. Wrap your model's `predict` call in `execute`. Since a UDF's `__init__` receives `validation_context` and `arguments` from the framework, override it to accept and forward both, then do your model loading after the `super().__init__()` call; for example: 267 267 + 268 268 + ```python 269 269 + class Arguments(ArgumentsBase): 270 270 + text: str 271 271 + 272 272 + class MySpamClassifier(UDFBase[Arguments, float]): 273 273 + def __init__(self, validation_context, arguments): 274 274 + super().__init__(validation_context, arguments) 275 275 + self._model = load_model("/path/to/model.pkl") 276 276 + 277 277 + def execute(self, execution_context: ExecutionContext, arguments: Arguments) -> float: 278 278 + return self._model.predict_proba([arguments.text])[0][1] 279 279 + ``` 280 280 + 281 281 + The returned score is then available in rules, e.g.: 282 282 + 283 283 + ```python 284 284 + MySpamClassifier(text=MessageContent) > 0.85 285 285 + ``` 286 286 + 287 287 + Osprey constructs one UDF instance per call site when the rules are compiled, not per event, so the model isn't reloaded for every action processed. Keep in mind this means per _call site_, not per _class_: if you call the same UDF from multiple rules, each call site gets its own instance, and each one loads its own copy of the model. **For a large model, prefer calling the UDF from a single rule (or share the loaded weights via a module-level cache) rather than invoking it from many places.** 288 288 + 289 289 + ## Packaging your plugin 290 290 + 291 291 + Your plugin package needs a `pyproject.toml` that declares the entry points; for example: 292 292 + 293 293 + ```toml 294 294 + [project] 295 295 + name = "my-osprey-plugins" 296 296 + version = "0.1.0" 297 297 + requires-python = ">=3.11" 298 298 + dependencies = ["pluggy==1.5.0"] 299 299 + 300 300 + [tool.setuptools] 301 301 + package-dir = {"" = "src"} 302 302 + 303 303 + [tool.setuptools.packages.find] 304 304 + where = ["src"] 305 305 + 306 306 + [project.entry-points.osprey_plugin] 307 307 + register_plugins = "register_plugins" 308 308 + ``` 309 309 + 310 310 + Install it into the same environment as Osprey and it will be discovered automatically on the next startup. 311 311 + 312 312 + See also: [Writing Rules](rules.md)