Wazuh ms-graph: why Entra ID sign-ins never reach the indexer, and a one-processor fix

You enable the Wazuh ms-graph module for auditLogs/signIns. The module logs clean scans. With wazuh_modules.debug=2 it even logs Sending log: {"integration":"ms-graph", ... "relationship":"signIns"}. And yet there is not a single sign-in in wazuh-alerts-* or wazuh-archives-*.

There are two separate reasons, and on a stock install you hit both.

1. The stock ruleset has no rule for sign-ins

0995-microsoft-graph_rules.xml in 4.14.7 has 99 rules. The parent rule 99500 (level 0) catches every ms-graph event. Its children cover alerts, incidents, audit events, devices, apps and risk detections. None matches relationship: signIns.

So on the stock ruleset a sign-in stops at level 0 and never becomes an alert. It can only reach the indexer through the archives (<logall_json>yes</logall_json>). That is where the second problem waits.

2. The stock template rejects every sign-in

wazuh-template.json applies to both wazuh-alerts-4.x-* and wazuh-archives-4.x-*. It maps two ms-graph fields as keyword:

field	template	alerts_v2 / incidents send	signIns send
data.ms-graph.status	keyword	a string ("resolved")	an object ({errorCode, failureReason, additionalDetails})
data.ms-graph.appliedConditionalAccessPolicies	keyword	not sent	a list of objects

A keyword field cannot hold an object, so the indexer rejects the document:

mapper_parsing_exception: failed to parse field [data.ms-graph.status] of type [keyword] ...
caused_by: illegal_state_exception: Can't get text on a START_OBJECT

Three details that make this harder to see than it should be:

• It does not depend on which event arrives first. "First shape wins" only holds for unmapped fields. These two are mapped explicitly, so every object-shaped sign-in is rejected, in any order, even if signIns is the only relationship you enabled. We checked the template at tags v4.8.0, v4.10.0, v4.12.0, v4.14.0, v4.14.3 and v4.14.7: the same in all six.
• The error names only the first bad field. A sign-in with at least one Conditional Access policy fails on appliedConditionalAccessPolicies. One with an empty policy list fails on status. Fix status alone and the sign-ins that had a policy applied are still rejected.
• Nothing on the manager says so. ossec.log is clean and archives.json has the event. The rejection happens inside the indexer, after Filebeat has shipped the event. One of the reporters found nothing in the Filebeat log either.

If the sign-in is missing from archives.json itself, you have a different problem, upstream of the indexer. Look at time_delay first: sign-in records show up in Graph with a delay, and a scan window that closes at "now" can miss them. Issue 39460 in wazuh/wazuh has both problems, one after the other.

Check yours in a minute

On the stock template timing does not matter, so the mapping tells you:

GET wazuh-archives-*,wazuh-alerts-*/_mapping/field/data.ms-graph.status,data.ms-graph.appliedConditionalAccessPolicies

If either field comes back "type": "keyword" and you have signIns enabled, every sign-in that reaches the indexer is rejected: through the archives, or through the alerts index once you add your own sign-in rule.

The fix: one script processor, by type

The idea is to leave the string status of alerts and incidents where it is, because dashboards and saved searches on data.ms-graph.status expect that string. Only the sign-in shapes move to new fields. The rules are not affected either way: they run on the manager, before the indexer.

• status, when it is an object → data.ms-graph.signInStatus
• appliedConditionalAccessPolicies, when it is a list of objects → data.ms-graph.conditionalAccessPolicies

It is one script processor (FIX in the script at the end), right after the json processor, in the alerts and the archives pipeline. Two details matter:

• ignore_failure: true. The stock Wazuh pipelines end with on_failure: drop, so a processor that throws does not just skip its own step; the whole event is dropped. With ignore_failure, the worst case is an event indexed unchanged.
• d.get('ms-graph'), not ctx.data?.ms-graph. Painless reads the hyphen as a minus sign, the dotted form does not compile, and the indexer refuses the whole pipeline.

The two new fields get a dynamic mapping on first use (signInStatus.errorCode becomes long, the text fields keyword). No template change is needed, and the fix works on today's index too; you do not wait for the next daily index.

The script at the end of this post does this for you, with a backup and a rollback:

export URL=https://<indexer>:9200 AUTH='<user>:<password>'
python3 msgraph_signins_fix.py verify     # 4 test events through your live pipelines
python3 msgraph_signins_fix.py apply      # saves both pipelines to ./backup, installs the processor
python3 msgraph_signins_fix.py verify     # expect 0 problems
python3 msgraph_signins_fix.py rollback   # restores ./backup

What we measured, and what we did not

Measured on a throwaway wazuh-indexer:4.14.7 with the stock template and both stock pipelines, taken from the wazuh-manager:4.14.7 image. We sent Graph v1.0-shaped sign-ins, with and without a Conditional Access policy, plus an alerts_v2 alert, the way Filebeat sends them:

• stock: every sign-in rejected, in both pipelines and in both arrival orders; the alert indexes fine
• status fixed alone: sign-ins that had a policy applied still rejected, now on the second field
• the processor: every sign-in and the alert indexed, in both orders, in a fresh index and in an index that already existed; data.ms-graph.status stays keyword and still aggregates; events without data are untouched
• the script: verify 4 errors → apply → verify 0 of 8 (including a non-ms-graph event) → a second apply skips → rollback restores both pipelines byte for byte → verify 4 errors again

Not measured: a real tenant, the Graph beta API (it adds sign-in fields we have not checked against the template), a real Filebeat in the path, and whether the patch survives a restart or upgrade. Two things to check on your side:

• If your filebeat.yml sets filebeat.overwrite_pipelines: true, Filebeat reloads the pipeline from disk and the patch has to go into the pipeline file instead.
• After a Wazuh upgrade, the pipeline id can change. Run verify again.

And to get sign-in alerts rather than archives only, you still need your own rule on ms-graph.relationship = signIns. The stock ruleset does not ship one.

Credit to the reporters of wazuh/wazuh issues 38330 and 39460: the first described the collision, the second narrowed the loss down to the indexer.

The script

Save it as msgraph_signins_fix.py. Standard library only; it changes nothing but the two pipelines, and verify writes only to indices dated 2099.01.01, then deletes them.

#!/usr/bin/env python3
# Wazuh ms-graph sign-in fix: verify | apply | rollback. ATK New Technology, vct.atkvn.com/#fix-pack
import base64, json, os, ssl, sys, urllib.error, urllib.request

URL, AUTH, BK = os.environ.get("URL", "").rstrip("/"), os.environ.get("AUTH", ""), "backup"
TAG = "ms-graph signIns fix"
SRC = ("def d = ctx.get('data'); if (!(d instanceof Map)) { return; } "
       "def g = d.get('ms-graph'); if (!(g instanceof Map)) { return; } "
       "if (g.get('status') instanceof Map) { g.put('signInStatus', g.remove('status')); } "
       "def p = g.get('appliedConditionalAccessPolicies'); "
       "if (p instanceof List && !p.isEmpty() && p.get(0) instanceof Map) "
       "{ g.put('conditionalAccessPolicies', g.remove('appliedConditionalAccessPolicies')); }")
FIX = {"script": {"lang": "painless", "source": SRC, "ignore_failure": True, "description": TAG}}


def es(method, path, body=None):
    req = urllib.request.Request(URL + path, method=method, headers={"Content-Type": "application/json"},
                                 data=None if body is None else json.dumps(body).encode())
    if AUTH:
        req.add_header("Authorization", "Basic " + base64.b64encode(AUTH.encode()).decode())
    try:  # stock Wazuh indexers use self-signed certificates
        with urllib.request.urlopen(req, context=ssl._create_unverified_context(), timeout=30) as r:
            return r.status, json.loads(r.read() or b"{}")
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read() or b"{}")


def pipelines():
    code, allp = es("GET", "/_ingest/pipeline")
    if code != 200:
        sys.exit("cannot list pipelines: HTTP %s" % code)
    out = {}
    for kind in ("alerts", "archives"):
        ids = sorted(p for p in allp if p.startswith("filebeat-") and p.endswith("-wazuh-%s-pipeline" % kind))
        if not ids:
            sys.exit("no filebeat-*-wazuh-%s-pipeline found" % kind)
        out[kind] = (ids[-1], allp[ids[-1]])
    return out


def apply():
    os.makedirs(BK, exist_ok=True)
    for kind, (pid, body) in pipelines().items():
        procs = body["processors"]
        if any(TAG in str(p.get("script", {}).get("description", "")) for p in procs):
            print(kind, pid, "already patched")
            continue
        i = next((i for i, p in enumerate(procs) if "json" in p), None)
        if i is None:
            sys.exit("%s has no json processor" % pid)
        path = os.path.join(BK, pid + ".json")
        if not os.path.exists(path):
            json.dump(body, open(path, "w"))
        code, r = es("PUT", "/_ingest/pipeline/" + pid, dict(body, processors=procs[:i + 1] + [FIX] + procs[i + 1:]))
        print(kind, pid, "HTTP", code, "backup", path)
        if code != 200:
            sys.exit(json.dumps(r)[:300])


def rollback():
    for kind, (pid, _) in pipelines().items():
        path = os.path.join(BK, pid + ".json")
        if os.path.exists(path):
            print(kind, pid, "restored, HTTP", es("PUT", "/_ingest/pipeline/" + pid, json.load(open(path)))[0])
        else:
            print(kind, pid, "no backup, left as is")


def verify():  # writes only to indices dated 2099.01.01, then deletes them
    ms = lambda **m: {"integration": "ms-graph", "ms-graph": m}
    st = {"errorCode": 50126, "failureReason": "Invalid username or password."}
    pol = [{"id": "0", "displayName": "verify", "result": "notApplied"}]
    cases = [("other event", {"srcip": "203.0.113.7"}),
             ("alerts_v2, string status", ms(status="resolved", relationship="alerts_v2")),
             ("signIns + policies", ms(status=st, appliedConditionalAccessPolicies=pol, relationship="signIns")),
             ("signIns, no policies", ms(status=st, appliedConditionalAccessPolicies=[], relationship="signIns"))]
    bad = 0
    for kind, (pid, _) in pipelines().items():
        pre = "wazuh-%s-4.x-" % kind
        for name, data in cases:
            a = {"timestamp": "2099-01-01T00:00:00.000+0000", "rule": {"id": "99500", "level": 0}, "data": data}
            code, r = es("POST", "/%sverify/_doc?pipeline=%s&refresh=true" % (pre, pid),
                         {"message": json.dumps(a), "fields": {"index_prefix": pre}})
            ok = r.get("result") == "created" and r.get("_index", "").endswith("2099.01.01")
            err = r.get("error")
            bad += not ok
            print("%-8s %-26s %s" % (kind, name, "OK" if ok else "ERROR %s" % (
                err.get("reason") if isinstance(err, dict) else r)[:160]))
        es("DELETE", "/%s2099.01.01" % pre)
    print(bad, "problem(s)")
    sys.exit(1 if bad else 0)


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else ""
    if not URL or cmd not in ("verify", "apply", "rollback"):
        sys.exit("usage: URL=https://<indexer>:9200 AUTH='<user>:<password>' %s verify|apply|rollback" % sys.argv[0])
    globals()[cmd]()

---

If you would rather not patch this yourself, we write and test the fix for your exact Wazuh version and send you the files: USD 490, paid only after it runs clean on your side. vct.atkvn.com/#fix-pack

Dong Nguyen, ATK New Technology