Skip to content

Apply all message filters to grouped triage #416

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
maximelb wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Apply all message filters to grouped triage #416

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions doc/cli/email-security.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -354,6 +354,21 @@ similar messages. Groups carry separate engine verdict/severity and analyst
disposition summaries. The default group list is the flagged triage queue; use
`--all` to include other groups. Aggregates expose an `as_of` timestamp.

Group listing accepts every message-list filter, with the same names and encoding:
`--search`/`--q`, `--mailbox`, `--sender-email`, `--sender-domain`/`--sender-root-domain`,
`--campaign-id`, `--group-id`, `--link-domain`, `--attachment-sha256`, repeatable
`--state`, `--direction`, `--verdict`, `--severity`, `--disposition` (including `none`),
`--min-score`, `--lane`, `--user-reported true|false`, `--since` and `--until`.
A group is returned only when **one recipient copy matches every active filter**.
Filtered order follows that group's newest matching copy; summary badges still
cover all copies. Search and lane have the same bounds/refusals as message lists.

Continue through short or empty pages while `next_cursor` is present. Filtered
cursors pin a snapshot for 50 minutes; restart on expiry or changing filters.
`group_filter_too_broad` asks for narrower selectors rather than dropping filters.
Exact `matched_copies` counts are omitted to bound per-page cost. Group actions
always act on **all copies frozen at preview**, including copies outside list filters.

```bash
limacharlie mailsec group list --severity high --severity critical --disposition none
limacharlie mailsec group get <group_id>
Expand Down
45 changes: 35 additions & 10 deletions limacharlie/commands/mailsec.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -931,19 +931,44 @@ def message_groups() -> None:
@click.option("--disposition", multiple=True, type=click.Choice(["malicious", "spam", "graymail", "benign", "simulation", "none"]))
@click.option("--user-reported", default=None, type=click.Choice(["true", "false"]))
@click.option("--all", "all_groups", is_flag=True, help="Include groups outside the flagged triage queue.")
@click.option("--since", default=None, help="Earliest last-seen time (RFC3339 or unix seconds).")
@click.option("--until", default=None, help="Exclusive latest last-seen time.")
@click.option("--since", default=None, help="Earliest matching-copy time (RFC3339 or unix seconds).")
@click.option("--until", default=None, help="Exclusive latest matching-copy time.")
@click.option("--cursor", default=None, help="Opaque cursor; keep all filters unchanged.")
@click.option("--limit", default=None, type=click.IntRange(1, 500), help="Page size.")
@click.option("--mailbox", default=None, help="Exact protected mailbox address.")
@click.option("--sender-email", default=None, help="Exact sender address.")
@click.option("--sender-domain", "--sender-root-domain", "sender_domain", default=None, help="Sender registrable root domain.")
@click.option("--campaign-id", default=None, help="Exact campaign identity.")
@click.option("--group-id", default=None, help="Exact message group identity.")
@click.option("--link-domain", default=None, help="Link registrable root domain.")
@click.option("--attachment-sha256", default=None, help="Attachment digest.")
@click.option("--state", multiple=True, help="Copy placement state (repeatable).")
@click.option("--direction", multiple=True, type=click.Choice(["inbound", "outbound", "internal"]))
@click.option("--min-score", default=None, type=click.IntRange(0,100), help="Minimum copy score.")
@click.option("--lane", default=None, type=click.Choice(["live", "backfill"]))
@click.option("--search", "--q", "q", default=None, help="Literal text search; requires --since or an indexed selector.")
@pass_context
def groups_list(ctx, verdict, severity, disposition, user_reported, all_groups, since, until, cursor, limit) -> None:
"""List the flagged-group triage queue, newest last-seen first."""
_output(ctx, _get_mailsec(ctx).list_groups(
verdict=list(verdict) or None, severity=list(severity) or None,
disposition=list(disposition) or None,
user_reported=None if user_reported is None else user_reported == "true",
all_groups=all_groups, since=since, until=until, cursor=cursor, limit=limit,
))
def groups_list(ctx, verdict, severity, disposition, user_reported, all_groups, since, until, cursor, limit, mailbox, sender_email, sender_domain, campaign_id, group_id, link_domain, attachment_sha256, state, direction, min_score, lane, q) -> None:
"""List groups whose one recipient copy matches every active filter.

Order uses the newest matching copy. Continue short or empty pages while a
next_cursor is present. Summaries and group actions cover ALL copies,
including copies outside the filters; exact matched-copy counts are omitted.
"""
try:
result = _get_mailsec(ctx).list_groups(
verdict=list(verdict) or None, severity=list(severity) or None,
disposition=list(disposition) or None,
user_reported=None if user_reported is None else user_reported == "true",
all_groups=all_groups, since=since, until=until, cursor=cursor, limit=limit,
mailbox=mailbox, sender_email=sender_email, sender_domain=sender_domain,
campaign_id=campaign_id, group_id=group_id, link_domain=link_domain,
attachment_sha256=attachment_sha256, state=list(state) or None,
direction=list(direction) or None, min_score=min_score, lane=lane, q=q,
)
except (TypeError, ValueError) as error:
raise click.ClickException(str(error)) from error
_output(ctx, result)


@message_groups.command("get")
Expand Down
195 changes: 144 additions & 51 deletions limacharlie/sdk/mailsec.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,69 @@ def get_coverage(
# Messages
# ------------------------------------------------------------------

@staticmethod
def _message_filter_pairs(
*,
verdict: list[str] | None = None,
severity: list[str] | None = None,
group_id: str | None = None,
mailbox: str | None = None,
sender_email: str | None = None,
sender_domain: str | None = None,
campaign_id: str | None = None,
state: list[str] | None = None,
direction: list[str] | None = None,
lane: str | None = None,
user_reported: bool | None = None,
min_score: int | None = None,
link_domain: str | None = None,
attachment_sha256: str | None = None,
q: str | None = None,
since: str | None = None,
until: str | None = None,
cursor: str | None = None,
limit: int | None = None,
disposition: list[str] | None = None,
) -> list[tuple[str, str]]:
if q is not None:
q = q.strip() or None
if q:
if len(q) > 512:
raise ValueError("q must be at most 512 code points")
bounded = any((since, mailbox, sender_email, campaign_id, group_id, link_domain, attachment_sha256))
single_verdict = verdict is not None and len(verdict) == 1 and bool(verdict[0].strip())
single_severity = severity is not None and len(severity) == 1 and bool(severity[0].strip())
if not bounded and not single_verdict and not single_severity:
raise ValueError(
"q requires since, mailbox, sender_email, campaign_id, "
"group_id, link_domain, attachment_sha256, or exactly one verdict/severity"
)
pairs: list[tuple[str, str]] = []
_add_pairs(pairs, "disposition", disposition)
_add_pairs(pairs, "verdict", verdict)
_add_pairs(pairs, "severity", severity)
_add_pairs(pairs, "state", state)
_add_pairs(pairs, "direction", direction)
for key, val in (
("mailbox", mailbox),
("sender_email", sender_email),
("sender_root_domain", sender_domain),
("campaign_id", campaign_id),
("group_id", group_id),
("min_score", min_score),
("link_domain", link_domain),
("attachment_sha256", attachment_sha256),
("q", q),
("since", since),
("until", until),
("cursor", cursor),
("limit", limit),
("user_reported", user_reported),
("lane", lane),
):
_add_scalar(pairs, key, val)
return pairs

def list_messages(
self,
*,
Expand Down Expand Up @@ -390,43 +453,28 @@ def list_messages(
"""
if disposition is not None and disposition not in (*DISPOSITIONS, "none"):
raise ValueError("invalid disposition filter")
if q is not None:
q = q.strip() or None
if q:
if len(q) > 512:
raise ValueError("q must be at most 512 code points")
bounded = any((since, mailbox, sender_email, campaign_id, group_id, link_domain, attachment_sha256))
single_verdict = verdict is not None and len(verdict) == 1 and bool(verdict[0].strip())
single_severity = severity is not None and len(severity) == 1 and bool(severity[0].strip())
if not bounded and not single_verdict and not single_severity:
raise ValueError(
"q requires since, mailbox, sender_email, campaign_id, "
"group_id, link_domain, attachment_sha256, or exactly one verdict/severity"
)
pairs: list[tuple[str, str]] = []
_add_pairs(pairs, "verdict", verdict)
_add_pairs(pairs, "severity", severity)
_add_pairs(pairs, "state", state)
_add_pairs(pairs, "direction", direction)
for key, val in (
("mailbox", mailbox),
("sender_email", sender_email),
("sender_root_domain", sender_domain),
("campaign_id", campaign_id),
("group_id", group_id),
("min_score", min_score),
("link_domain", link_domain),
("attachment_sha256", attachment_sha256),
("q", q),
("since", since),
("until", until),
("cursor", cursor),
("limit", limit),
("user_reported", user_reported),
("lane", lane),
("disposition", disposition),
):
_add_scalar(pairs, key, val)
pairs = self._message_filter_pairs(
disposition=[disposition] if disposition is not None else None,
verdict=verdict,
severity=severity,
group_id=group_id,
mailbox=mailbox,
sender_email=sender_email,
sender_domain=sender_domain,
campaign_id=campaign_id,
state=state,
direction=direction,
lane=lane,
user_reported=user_reported,
min_score=min_score,
link_domain=link_domain,
attachment_sha256=attachment_sha256,
q=q,
since=since,
until=until,
cursor=cursor,
limit=limit,
)
return self._get("messages", pairs)

def get_message(self, msg_uuid: str) -> dict[str, Any]:
Expand Down Expand Up @@ -970,33 +1018,78 @@ def list_groups(
disposition: list[str] | None = None, user_reported: bool | None = None,
all_groups: bool = False, since: str | None = None, until: str | None = None,
cursor: str | None = None, limit: int | None = None,
mailbox: str | None = None, sender_email: str | None = None,
sender_domain: str | None = None, campaign_id: str | None = None,
group_id: str | None = None, link_domain: str | None = None,
attachment_sha256: str | None = None, state: list[str] | None = None,
direction: list[str] | None = None, min_score: int | None = None,
lane: str | None = None, q: str | None = None,
) -> dict[str, Any]:
"""List the flagged message-group queue, ordered by last seen.
"""List groups whose ONE recipient copy matches ALL active filters.

Filters use the same encoding and meanings as :meth:`list_messages`:
alternatives within a key, AND across keys on one copy. Filtered order
is the newest matching copy. Returned summaries and group actions always
cover the whole group; exact matched-copy counts are omitted to bound cost.
Continue through short/empty pages until next_cursor is empty. Cursors
pin a snapshot for 50 minutes; restart after expiry or changing filters.

Args:
verdict: Repeatable engine verdict filter.
severity: Repeatable severity filter.
disposition: Repeatable analyst disposition filter, including none.
severity: Repeatable rule-impact filter.
disposition: Repeatable analyst disposition, including none.
user_reported: True or false to constrain reports; None is unconstrained.
all_groups: Include groups outside the flagged queue.
since: Earliest last-seen time, RFC3339 or unix seconds.
until: Exclusive latest last-seen time.
cursor: Opaque cursor bound to the complete filter set.
limit: Page size.
since: Inclusive matching-copy time, RFC3339 or unix seconds.
until: Exclusive matching-copy time.
cursor: Opaque filter-bound snapshot cursor.
limit: Maximum returned groups per page.
mailbox: Exact protected mailbox address.
sender_email: Exact sender address.
sender_domain: Sender registrable domain (sender_root_domain on the wire).
campaign_id: Exact campaign identity.
group_id: Exact message-group identity.
link_domain: Link registrable domain.
attachment_sha256: Attachment digest.
state: Repeatable copy placement state.
direction: Repeatable inbound, outbound or internal direction.
min_score: Minimum copy score.
lane: Live or backfill, with the same supported combinations as messages.
q: Literal case-insensitive subject/sender substring; requires a bounded-walk companion.

Returns:
dict: Groups, next_cursor and materialized as_of timestamps.
dict: Whole-group summaries, next_cursor and as_of timestamps.

Raises:
TypeError: If all_groups is not boolean.
ValueError: If non-empty q is too long or lacks a bounded-walk companion.
"""
pairs: list[tuple[str, str]] = []
for key, values in (("verdict", verdict), ("severity", severity), ("disposition", disposition)):
_add_pairs(pairs, key, values)
_add_scalar(pairs, "user_reported", user_reported)
if not isinstance(all_groups, bool):
raise TypeError("all_groups must be a boolean")
pairs = self._message_filter_pairs(
verdict=verdict,
severity=severity,
group_id=group_id,
mailbox=mailbox,
sender_email=sender_email,
sender_domain=sender_domain,
campaign_id=campaign_id,
state=state,
direction=direction,
lane=lane,
user_reported=user_reported,
min_score=min_score,
link_domain=link_domain,
attachment_sha256=attachment_sha256,
q=q,
since=since,
until=until,
cursor=cursor,
limit=limit,
disposition=disposition,
)
if all_groups:
_add_scalar(pairs, "all", True)
for key, value in (("since", since), ("until", until), ("cursor", cursor), ("limit", limit)):
_add_scalar(pairs, key, value)
return self._get("groups", pairs)

def get_group(self, group_id: str) -> dict[str, Any]:
Expand Down
19 changes: 19 additions & 0 deletions tests/unit/test_cli_mailsec_groups.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,3 +76,22 @@ def test_group_disposition_preview_freezes_note_and_clear_without_execution():
result, ms = invoke("preview", GID, "--action", "set_disposition", "--disposition", "true_positive")
assert result.exit_code != 0
ms.prepare_group_action.assert_not_called()


def test_group_list_forwards_every_copy_filter():
result, ms = invoke("list", "--mailbox", "copy@example.invalid", "--sender-email", "sender@example.invalid",
"--sender-root-domain", "example.invalid", "--campaign-id", JOB, "--group-id", GID,
"--link-domain", "linked.invalid", "--attachment-sha256", "b" * 64,
"--state", "delivered", "--state", "quarantined", "--direction", "inbound",
"--direction", "internal", "--min-score", "70", "--lane", "live", "--q", "literal 50%",
"--since", "2026-10-01T00:00:00Z", "--until", "2026-10-02T00:00:00Z",
"--verdict", "malicious", "--severity", "critical", "--disposition", "none",
"--user-reported", "false", "--all", "--cursor", "opaque", "--limit", "5")
assert result.exit_code == 0, result.output
ms.list_groups.assert_called_once_with(
verdict=["malicious"], severity=["critical"], disposition=["none"], user_reported=False, all_groups=True,
since="2026-10-01T00:00:00Z", until="2026-10-02T00:00:00Z", cursor="opaque", limit=5,
mailbox="copy@example.invalid", sender_email="sender@example.invalid", sender_domain="example.invalid",
campaign_id=JOB, group_id=GID, link_domain="linked.invalid", attachment_sha256="b" * 64,
state=["delivered", "quarantined"], direction=["inbound", "internal"], min_score=70, lane="live", q="literal 50%",
)
26 changes: 26 additions & 0 deletions tests/unit/test_sdk_mailsec_groups.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,3 +128,29 @@ def test_invalid_group_disposition_refused_before_transport(client, action, para
with pytest.raises(ValueError):
ms.prepare_group_action(GID, action, JOB, **params)
transport.request.assert_not_called()


def test_group_and_message_filter_wire_parity(client):
ms, transport = client
filters = dict(verdict=["malicious", "suspicious"], severity=["high", "critical"],
group_id=GID, mailbox="copy@example.invalid", sender_email="sender@example.invalid",
sender_domain="example.invalid", campaign_id=JOB, state=["delivered", "quarantined"],
direction=["inbound", "internal"], lane="live", user_reported=False, min_score=70,
link_domain="linked.invalid", attachment_sha256="b" * 64, q=" literal 50% ",
since="2026-10-01T00:00:00Z", until="2026-10-02T00:00:00Z", cursor="opaque", limit=5)
ms.list_messages(**filters)
message_pairs = transport.request.call_args.kwargs["query_params"]
ms.list_groups(**filters, disposition=["malicious", "none"], all_groups=True)
group_pairs = transport.request.call_args.kwargs["query_params"]
assert [(k, v) for k, v in group_pairs if k not in ("disposition", "all")] == message_pairs
assert ("disposition", "none") in group_pairs
assert ("q", "literal 50%") in group_pairs


@pytest.mark.parametrize("q,kwargs", [("x" * 513, {"since": "2026-10-01"}), ("needle", {}),
("needle", {"until": "2026-10-02"})])
def test_group_search_refuses_unbounded_or_oversized_input(client, q, kwargs):
ms, transport = client
with pytest.raises(ValueError):
ms.list_groups(q=q, **kwargs)
transport.request.assert_not_called()
Loading