>I wrote a small Python normalizer. It flattens the key indicators and metadata we care about
You're on the right track, but you've built a ticking time bomb. Flattening for ingestion is a necessary evil, but you haven't insulated your playbooks from the source. Your function `normalize_mandiant_item` is your single point of failure tied directly to their v4 schema.
In our deployment, we learned this the hard way. We built a similar normalizer, and a minor field rename from `industries_targeted` to `targeted_industries` in the Mandiant feed broke seven downstream playbooks that were looking for the exact flattened key. The script ran fine, output a field, but the playbooks were waiting for a ghost.
You need a translation layer *before* the flattening. Don't output `industries_targeted`. Output your own canonical field name, like `target_sectors`. Your internal schema should be owned by you. Then your mapping logic absorbs the vendor's changes. When they rename the field, you update one line in your mapping dictionary, not every playbook.
Also, you mentioned cleaning list-of-dicts into arrays. How are you handling the inevitable day when one of those dicts contains a crucial nested value you didn't account for? Your flattening will discard it silently. You need to validate the output schema against a defined profile for each object type, or you're just trading one form of inconsistency for another.
The mapping dictionary is the right call, but you're still trusting the vendor's field *existence*. The real kicker is when a crucial nested field you've been extracting just disappears from their schema, or shows up null for the first time. Your internal `target_sectors` becomes an empty list, and your playbook logic built around its presence might still run, just incorrectly.
The dictionary should map to a *function*, not just a string key. That function can handle the extraction, provide a default if the path is gone, and log the schema drift for your time-tracking spreadsheet. It's more code upfront, but it turns a pipeline break into a logged warning.
Yeah, mapping to a function is the right move for handling those edge cases. We do something similar and added a simple health check that runs weekly - it logs a warning if any of our extractor functions start returning default values more than, say, 10% of the time. That way you get a heads-up about gradual schema drift before it silently breaks something.
It does add complexity, but it beats getting paged at 3 AM because a playbook made a wrong decision on empty data.
Automate everything.
The weekly health check on default values is a smart extension, but your 10% threshold might be too permissive for critical decision fields. For something like `ttp_list` or `c2_ips`, even a 5% default rate indicates the source schema is no longer reliable and the playbook's logic is now operating on potentially false negatives. That threshold should be field-dependent.
You also need to track the *variance* of those default rates. A sudden jump from 0% to 10% is a critical alert, while a slow creep from 2% to 10% over months is a product management failure. Logging just the weekly percentage misses that trend data. Store the raw counts and timestamps; you can graph it later when arguing for a vendor fix.
Thanks for sharing this, it looks really helpful for getting started. I'm just beginning to work with external feeds like this in our SOAR.
A quick question about the main function - when you say it handles malware, actors, and vulnerabilities, does that mean you run it separately for each type, or does it figure out the object type automatically from the JSON? I'm trying to picture the workflow.
Good question. The original script likely figures out the object type automatically from a top-level field in the JSON, something like `type` or `object_type`. You'd typically run the same normalizer function over each item in the feed, and it handles the branching internally. This way you can process a mixed batch of indicators.
Just watch out for any new object types the vendor might add in the future. If your normalizer only checks for the three types you know, a new fourth type could get silently dropped or cause an error. Adding a default case that logs an unexpected type is a simple safeguard.
—HR
Exactly, that top-level type field is the pivot. The branching is the easy part though. The real problem is when the vendor adds a new object type *and* that type has a completely different internal structure for common fields you're already extracting. Your "malware" type might have `ttp_list` as an array of strings, but the new "campaign" type might nest it under `behavior.techniques` as an array of objects.
You can't just catch the new type and log it. You have to decide if you're going to attempt to map it using your existing field extractors, which will likely fail or produce garbage, or treat it as a wholly new schema requiring a separate mapping dictionary. We built a secondary dispatcher that checks if the new object type's JSON structure is congruent with one of our known patterns before even trying the standard extraction functions. If it isn't, it routes to a quarantine bucket for manual review. This stops corrupted data from polluting your pipeline.