diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index 3d1854df..cd219530 100644 --- a/.github/workflows/codeql.yml +++ b/.github/workflows/codeql.yml @@ -24,18 +24,18 @@ jobs: steps: - name: Checkout - uses: actions/checkout@v3 + uses: actions/checkout@v4 - name: Initialize CodeQL - uses: github/codeql-action/init@v2 + uses: github/codeql-action/init@v3 with: languages: ${{ matrix.language }} queries: +security-and-quality - name: Autobuild - uses: github/codeql-action/autobuild@v2 + uses: github/codeql-action/autobuild@v3 - name: Perform CodeQL Analysis - uses: github/codeql-action/analyze@v2 + uses: github/codeql-action/analyze@v3 with: category: "/language:${{ matrix.language }}" diff --git a/README.md b/README.md index 41de1a35..0f496119 100644 --- a/README.md +++ b/README.md @@ -29,3 +29,39 @@ In order for this watcher to be available in the UI, you'll need to have a Away To log current window title the terminal needs access to macOS accessibility API. This can be enabled in `System Preferences > Security & Privacy > Accessibility`, then add the Terminal to this list. If this is not enabled the watcher can only log current application, and not window title. +## Privacy Filter + +You can configure rules to **drop** or **redact** sensitive window events before they are sent to aw-server. This is a client-side pre-filter: matching events never leave the machine at all (on the Python heartbeat path; see the macOS note below). + +Add `[[aw-watcher-window.privacy_filter]]` entries to your config file +(`~/.config/activitywatch/aw-watcher-window/aw-watcher-window.toml`): + +```toml +# Drop events from private-browsing windows entirely +[[aw-watcher-window.privacy_filter]] +pattern = "(?i)private browsing|incognito" +action = "drop" + +# Redact window titles that contain sensitive account information +[[aw-watcher-window.privacy_filter]] +pattern = "(?i)bank|my account|password" +action = "redact" +replacement = "REDACTED" # optional; defaults to "excluded" + +# Drop events from a specific app by matching its name +[[aw-watcher-window.privacy_filter]] +pattern = "(?i)signal|whatsapp" +field = "app" # optional; defaults to "title" +action = "drop" +``` + +Rule fields: +- `pattern` — Python `re` regex (case-insensitive flag supported via `(?i)`) +- `field` — window-event field to match against (`"title"` by default; use `"app"` to match on application name) +- `action` — `"drop"` (discard the event) or `"redact"` (replace the field value) +- `replacement` — string to use when redacting; defaults to `"excluded"` + +Rules are applied in order. A `"drop"` rule exits immediately — subsequent rules are not evaluated for that event. + +> **macOS note**: The default `swift` strategy sends heartbeats from a separate binary and cannot apply these rules. If `privacy_filter` is configured, the watcher **refuses to start** under `--strategy swift` so titles are not leaked. Use `--strategy jxa` or `--strategy applescript` on macOS. + diff --git a/aw_watcher_window/config.py b/aw_watcher_window/config.py index e6bf075f..6891d95d 100644 --- a/aw_watcher_window/config.py +++ b/aw_watcher_window/config.py @@ -1,4 +1,5 @@ import argparse +from collections.abc import Mapping import tomlkit @@ -94,4 +95,10 @@ def parse_args(): parsed_args = parser.parse_args() parsed_args.research_category_map = dict(config.get("research_category_map", {})) parsed_args.research_app_category_map = dict(config.get("research_app_category_map", {})) + privacy_filter_cfg = config.get("privacy_filter", []) + # A single [aw-watcher-window.privacy_filter] table is a common TOML + # mistake; list(mapping) would yield the keys and silently drop the rule. + if isinstance(privacy_filter_cfg, Mapping): + privacy_filter_cfg = [privacy_filter_cfg] + parsed_args.privacy_filter_rules = list(privacy_filter_cfg or []) return parsed_args diff --git a/aw_watcher_window/main.py b/aw_watcher_window/main.py index 9d20740b..e5c9503e 100644 --- a/aw_watcher_window/main.py +++ b/aw_watcher_window/main.py @@ -14,6 +14,7 @@ from .config import parse_args from .exceptions import FatalError from .lib import get_current_window +from .privacy_filter import apply_privacy_filters, compile_privacy_rules from .research_filter import transform as research_transform from .macos_cli import build_swift_command from .macos_permissions import background_ensure_permissions @@ -95,7 +96,18 @@ def main(): if args.research_enabled else None ) + privacy_filter_rules = compile_privacy_rules( + getattr(args, "privacy_filter_rules", []) + ) if sys.platform == "darwin" and args.strategy == "swift": + if privacy_filter_rules: + logger.error( + "privacy_filter is configured, but the macOS swift strategy " + "sends heartbeats from a separate binary and cannot apply " + "those rules. Refusing to start so window titles are not " + "leaked. Use --strategy jxa or --strategy applescript." + ) + sys.exit(1) logger.info("Using swift strategy, calling out to swift binary") binpath = os.path.join( os.path.dirname(os.path.realpath(__file__)), "aw-watcher-window-macos" @@ -135,6 +147,7 @@ def main(): ], research_category_map=research_category_map, research_app_category_map=research_app_category_map, + privacy_filter_rules=privacy_filter_rules, ) @@ -147,6 +160,7 @@ def heartbeat_loop( exclude_titles=[], research_category_map=None, research_app_category_map=None, + privacy_filter_rules=None, ): while True: if os.getppid() == 1: @@ -186,17 +200,21 @@ def heartbeat_loop( exclude_titles=exclude_titles, research_category_map=research_category_map, research_app_category_map=research_app_category_map, + privacy_filter_rules=privacy_filter_rules, ) - now = datetime.now(timezone.utc) - current_window_event = Event(timestamp=now, data=current_window) - - client.heartbeat( - bucket_id, - current_window_event, - pulsetime=compute_pulsetime(poll_time), - queued=True, - ) + if current_window is None: + logger.debug("Event dropped by privacy filter, skipping heartbeat") + else: + now = datetime.now(timezone.utc) + current_window_event = Event(timestamp=now, data=current_window) + + client.heartbeat( + bucket_id, + current_window_event, + pulsetime=compute_pulsetime(poll_time), + queued=True, + ) sleep(poll_time) @@ -207,7 +225,14 @@ def transform_window( exclude_titles=None, research_category_map=None, research_app_category_map=None, + privacy_filter_rules=None, ): + # Privacy filter runs first — sensitive events never reach other transforms + if privacy_filter_rules: + current_window = apply_privacy_filters(current_window, privacy_filter_rules) + if current_window is None: + return None + if research_category_map is not None: return research_transform( current_window, diff --git a/aw_watcher_window/privacy_filter.py b/aw_watcher_window/privacy_filter.py new file mode 100644 index 00000000..1ce4deec --- /dev/null +++ b/aw_watcher_window/privacy_filter.py @@ -0,0 +1,133 @@ +""" +Watcher-side privacy filter: drop or redact sensitive events before sending. + +Mirrors the server-side ``privacy_filters`` engine in aw-server-rust so that +sensitive data never leaves the machine in the first place. + +Configure rules in ``~/.config/activitywatch/aw-watcher-window/aw-watcher-window.toml``: + +.. code-block:: toml + + # Drop events whose title matches a private-browsing pattern + [[aw-watcher-window.privacy_filter]] + pattern = "(?i)private browsing|incognito" + action = "drop" + + # Redact window titles containing a banking domain + [[aw-watcher-window.privacy_filter]] + pattern = "(?i)bank\\.example\\.com" + action = "redact" + replacement = "REDACTED" # optional; defaults to "excluded" + +Rule fields: + pattern -- Python ``re`` regex applied to the target field value. + field -- Window-event field to match against (default: ``"title"``). + action -- ``"drop"`` (discard the event) or ``"redact"`` (replace the value). + replacement -- Replacement string for ``"redact"`` action (default: ``"excluded"``). + +macOS note: the default ``swift`` strategy cannot apply these rules (it +sends heartbeats from a separate binary). The watcher refuses to start +under ``--strategy swift`` when any rule compiled, so titles are not +leaked. Use ``--strategy jxa`` or ``--strategy applescript`` on macOS. +""" + +import logging +import re +from typing import Optional + +logger = logging.getLogger(__name__) + +_VALID_ACTIONS = frozenset({"drop", "redact"}) + + +def compile_privacy_rules(raw_rules: list) -> list: + """Validate and compile regex patterns in privacy filter rules. + + Returns a list of compiled rule dicts ready for :func:`apply_privacy_filters`. + Invalid rules (bad regex or unknown action) are skipped with an error log. + """ + compiled = [] + for raw in raw_rules: + if not isinstance(raw, dict): + logger.error("privacy_filter rule is not a table: %r — skipped", raw) + continue + + pattern_str = raw.get("pattern", "") + if not pattern_str: + logger.error("privacy_filter rule missing 'pattern' — skipped: %r", raw) + continue + + try: + pattern = re.compile(pattern_str) + except (re.error, TypeError) as exc: + logger.error( + "privacy_filter: invalid regex %r — %s — rule skipped", pattern_str, exc + ) + continue + + action = raw.get("action", "redact") + if action not in _VALID_ACTIONS: + logger.error( + "privacy_filter: unknown action %r (expected 'drop' or 'redact') — rule skipped", + action, + ) + continue + + field = raw.get("field", "title") + if not isinstance(field, str) or not field: + logger.error( + "privacy_filter: 'field' must be a non-empty string — rule skipped: %r", + raw, + ) + continue + + replacement = raw.get("replacement", "excluded") + if not isinstance(replacement, str): + logger.error( + "privacy_filter: 'replacement' must be a string — rule skipped: %r", + raw, + ) + continue + + compiled.append( + { + "pattern": pattern, + "field": field, + "action": action, + "replacement": replacement, + } + ) + return compiled + + +def apply_privacy_filters(window: dict, rules: list) -> Optional[dict]: + """Apply compiled privacy filter rules to a window event. + + Returns ``None`` when the event should be dropped entirely, otherwise a + (possibly modified) copy of the window dict. Never mutates the input. + """ + if not rules: + return window + + result = dict(window) + for rule in rules: + field = rule["field"] + value = result.get(field) + if not isinstance(value, str): + continue + if rule["pattern"].search(value): + if rule["action"] == "drop": + logger.debug( + "privacy_filter: dropping event (field=%r matched %r)", + field, + rule["pattern"].pattern, + ) + return None + else: # redact + logger.debug( + "privacy_filter: redacting field=%r matched %r", + field, + rule["pattern"].pattern, + ) + result[field] = rule["replacement"] + return result diff --git a/tests/test_config.py b/tests/test_config.py index 668104a9..f1de0fca 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -105,6 +105,59 @@ def test_parse_args_defaults_research_off_without_config(tmp_path, monkeypatch): assert args.research_app_category_map == {} +def test_parse_args_wraps_single_privacy_filter_table(monkeypatch): + """A single [privacy_filter] table must not be list()'d into its keys.""" + monkeypatch.setattr( + config_module, + "load_config", + lambda: { + "exclude_title": False, + "exclude_titles": [], + "poll_time": 1.0, + "strategy_macos": "swift", + "privacy_filter": { + "pattern": "(?i)bank", + "action": "redact", + "replacement": "REDACTED", + }, + }, + ) + monkeypatch.setattr(sys, "argv", ["aw-watcher-window"]) + + args = config_module.parse_args() + + assert args.privacy_filter_rules == [ + { + "pattern": "(?i)bank", + "action": "redact", + "replacement": "REDACTED", + } + ] + + +def test_parse_args_keeps_privacy_filter_array(monkeypatch): + monkeypatch.setattr( + config_module, + "load_config", + lambda: { + "exclude_title": False, + "exclude_titles": [], + "poll_time": 1.0, + "strategy_macos": "swift", + "privacy_filter": [ + {"pattern": "incognito", "action": "drop"}, + {"pattern": "bank", "action": "redact"}, + ], + }, + ) + monkeypatch.setattr(sys, "argv", ["aw-watcher-window"]) + + args = config_module.parse_args() + + assert len(args.privacy_filter_rules) == 2 + assert args.privacy_filter_rules[0]["action"] == "drop" + + def test_parse_args_attaches_research_category_map(monkeypatch): monkeypatch.setattr( config_module, diff --git a/tests/test_main.py b/tests/test_main.py index facda112..94e7270b 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -200,6 +200,110 @@ def test_legacy_exclude_titles_still_apply_without_research_mode(): assert transformed == {"app": "Chrome", "title": "excluded"} +def test_exclude_titles_apply_when_privacy_rules_do_not_match(): + """Privacy filter copies the window; exclude_titles must mutate that copy. + + Regression for a reviewer claim that exclude_titles is lost whenever any + privacy_filter rule is configured, even if it does not match. + """ + from aw_watcher_window.privacy_filter import compile_privacy_rules + + window = {"app": "Chrome", "title": "secret document"} + rules = compile_privacy_rules([{"pattern": "bank", "action": "redact"}]) + + transformed = main_module.transform_window( + window, + exclude_titles=[re.compile("secret", re.IGNORECASE)], + privacy_filter_rules=rules, + ) + + assert transformed == {"app": "Chrome", "title": "excluded"} + assert window["title"] == "secret document" + + +def test_privacy_redact_then_exclude_titles_sees_redacted_copy(): + from aw_watcher_window.privacy_filter import compile_privacy_rules + + window = {"app": "Chrome", "title": "my bank dashboard"} + rules = compile_privacy_rules( + [{"pattern": "bank", "action": "redact", "replacement": "REDACTED"}] + ) + + transformed = main_module.transform_window( + window, + exclude_titles=[re.compile("bank", re.IGNORECASE)], + privacy_filter_rules=rules, + ) + + # exclude_titles runs on the redacted copy, so "bank" no longer matches + assert transformed == {"app": "Chrome", "title": "REDACTED"} + assert window["title"] == "my bank dashboard" + + +def test_swift_refuses_to_start_when_privacy_filter_configured(monkeypatch): + commands = [] + + class FakeProcess: + pid = 123 + + def wait(self): + return None + + class FakeClient: + client_name = "aw-watcher-window" + client_hostname = "host.localdomain" + server_address = "http://localhost:5600" + + def __init__(self, *args, **kwargs): + pass + + def create_bucket(self, *args, **kwargs): + pass + + def wait_for_start(self): + pass + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + monkeypatch.setattr(main_module.sys, "platform", "darwin") + monkeypatch.setattr(main_module, "background_ensure_permissions", lambda: None) + monkeypatch.setattr(main_module, "setup_logging", lambda **kwargs: None) + monkeypatch.setattr(main_module, "ActivityWatchClient", FakeClient) + monkeypatch.setattr(main_module.signal, "signal", lambda *args, **kwargs: None) + monkeypatch.setattr( + main_module.subprocess, + "Popen", + lambda command: commands.append(command) or FakeProcess(), + ) + monkeypatch.setattr( + main_module, + "parse_args", + lambda: SimpleNamespace( + testing=True, + verbose=False, + host=None, + port=None, + strategy="swift", + exclude_title=False, + exclude_titles=[], + research_enabled=False, + research_category_map={}, + research_app_category_map={}, + privacy_filter_rules=[{"pattern": "(?i)bank", "action": "drop"}], + ), + ) + + with pytest.raises(SystemExit) as excinfo: + main_module.main() + + assert excinfo.value.code == 1 + assert commands == [] + + @pytest.mark.parametrize( "poll_time,expected_pulsetime", [ diff --git a/tests/test_privacy_filter.py b/tests/test_privacy_filter.py new file mode 100644 index 00000000..fb7f514a --- /dev/null +++ b/tests/test_privacy_filter.py @@ -0,0 +1,209 @@ +"""Tests for the watcher-side privacy filter.""" + +import pytest + +from aw_watcher_window.privacy_filter import apply_privacy_filters, compile_privacy_rules + + +# --------------------------------------------------------------------------- +# compile_privacy_rules +# --------------------------------------------------------------------------- + + +def test_compile_empty_rules(): + assert compile_privacy_rules([]) == [] + + +def test_compile_valid_drop_rule(): + rules = compile_privacy_rules( + [{"pattern": "(?i)incognito", "action": "drop"}] + ) + assert len(rules) == 1 + assert rules[0]["action"] == "drop" + assert rules[0]["field"] == "title" + assert rules[0]["replacement"] == "excluded" + + +def test_compile_valid_redact_rule(): + rules = compile_privacy_rules( + [{"pattern": "bank", "action": "redact", "replacement": "REDACTED", "field": "title"}] + ) + assert len(rules) == 1 + assert rules[0]["replacement"] == "REDACTED" + + +def test_compile_defaults_to_redact(): + rules = compile_privacy_rules([{"pattern": "secret"}]) + assert rules[0]["action"] == "redact" + + +def test_compile_invalid_regex_skipped(): + rules = compile_privacy_rules([{"pattern": "[invalid(", "action": "drop"}]) + assert rules == [] + + +def test_compile_non_string_pattern_skipped(): + rules = compile_privacy_rules([{"pattern": 123, "action": "drop"}]) + assert rules == [] + + +def test_compile_unknown_action_skipped(): + rules = compile_privacy_rules([{"pattern": "foo", "action": "transform"}]) + assert rules == [] + + +def test_compile_missing_pattern_skipped(): + rules = compile_privacy_rules([{"action": "drop"}]) + assert rules == [] + + +def test_compile_non_dict_skipped(): + rules = compile_privacy_rules(["not a dict"]) + assert rules == [] + + +def test_compile_non_string_field_skipped(): + rules = compile_privacy_rules( + [{"pattern": "bank", "action": "redact", "field": 123}] + ) + assert rules == [] + + +def test_compile_non_string_replacement_skipped(): + rules = compile_privacy_rules( + [{"pattern": "bank", "action": "redact", "replacement": 123}] + ) + assert rules == [] + + +def test_compile_multiple_rules(): + raw = [ + {"pattern": "incognito", "action": "drop"}, + {"pattern": "bank", "action": "redact"}, + ] + compiled = compile_privacy_rules(raw) + assert len(compiled) == 2 + + +# --------------------------------------------------------------------------- +# apply_privacy_filters +# --------------------------------------------------------------------------- + + +@pytest.fixture +def drop_rule(): + return compile_privacy_rules( + [{"pattern": "(?i)private browsing|incognito", "action": "drop"}] + ) + + +@pytest.fixture +def redact_rule(): + return compile_privacy_rules( + [{"pattern": "(?i)bank", "action": "redact", "replacement": "REDACTED"}] + ) + + +def test_no_rules_passes_through(): + window = {"app": "Firefox", "title": "Incognito Window"} + assert apply_privacy_filters(window, []) == window + + +def test_drop_matching_event(drop_rule): + window = {"app": "Firefox", "title": "Private Browsing - Mozilla Firefox"} + result = apply_privacy_filters(window, drop_rule) + assert result is None + + +def test_drop_case_insensitive(drop_rule): + window = {"app": "Chrome", "title": "incognito"} + result = apply_privacy_filters(window, drop_rule) + assert result is None + + +def test_drop_non_matching_passes_through(drop_rule): + window = {"app": "Firefox", "title": "Hacker News"} + result = apply_privacy_filters(window, drop_rule) + assert result == window + + +def test_redact_matching_title(redact_rule): + window = {"app": "Chrome", "title": "My Bank - Dashboard"} + result = apply_privacy_filters(window, redact_rule) + assert result is not None + assert result["title"] == "REDACTED" + assert result["app"] == "Chrome" + + +def test_redact_default_replacement(): + rules = compile_privacy_rules([{"pattern": "secret", "action": "redact"}]) + window = {"app": "Terminal", "title": "secret notes"} + result = apply_privacy_filters(window, rules) + assert result["title"] == "excluded" + + +def test_redact_does_not_mutate_input(redact_rule): + window = {"app": "Chrome", "title": "bankofamerica.com"} + original_title = window["title"] + apply_privacy_filters(window, redact_rule) + assert window["title"] == original_title + + +def test_drop_does_not_mutate_input(drop_rule): + window = {"app": "Firefox", "title": "Private Browsing"} + original = dict(window) + apply_privacy_filters(window, drop_rule) + assert window == original + + +def test_field_not_present_skips_rule(redact_rule): + window = {"app": "Terminal"} # no 'title' field + result = apply_privacy_filters(window, redact_rule) + assert result == window + + +def test_non_string_field_skipped(): + rules = compile_privacy_rules([{"pattern": "123", "action": "drop", "field": "count"}]) + window = {"app": "Foo", "title": "bar", "count": 123} + result = apply_privacy_filters(window, rules) + assert result is not None # non-string field value: rule skipped + + +def test_multiple_rules_first_drop_wins(): + rules = compile_privacy_rules( + [ + {"pattern": "drop_me", "action": "drop"}, + {"pattern": "drop_me", "action": "redact"}, + ] + ) + window = {"app": "App", "title": "drop_me now"} + assert apply_privacy_filters(window, rules) is None + + +def test_multiple_rules_redact_then_drop(): + rules = compile_privacy_rules( + [ + {"pattern": "bank", "action": "redact", "replacement": "REDACTED"}, + {"pattern": "REDACTED", "action": "drop"}, + ] + ) + window = {"app": "Chrome", "title": "my bank account"} + # First rule redacts "bank" → "REDACTED"; second drops "REDACTED" + assert apply_privacy_filters(window, rules) is None + + +def test_custom_field(): + rules = compile_privacy_rules( + [{"pattern": "(?i)messenger", "action": "drop", "field": "app"}] + ) + window = {"app": "Facebook Messenger", "title": "Chat"} + assert apply_privacy_filters(window, rules) is None + + +def test_custom_field_does_not_affect_title(): + rules = compile_privacy_rules( + [{"pattern": "(?i)messenger", "action": "drop", "field": "app"}] + ) + window = {"app": "Terminal", "title": "messenger in title"} + # app doesn't match; rule scoped to 'app' field + assert apply_privacy_filters(window, rules) == window