Merge pull request #1667 from doronz88/feature/syslog-no-debug-info

syslog: Add `--no-debug` and `--no-info`

Author: doronz88

pymobiledevice3/cli/syslog.py

@@ -19,7 +19,13 @@
     user_requested_colored_output,
 )
 from pymobiledevice3.lockdown_service_provider import LockdownServiceProvider
-from pymobiledevice3.services.os_trace import OsTraceService, SyslogEntry, SyslogLogLevel
+from pymobiledevice3.services.os_trace import (
+    OS_TRACE_RELAY_STREAM_FLAGS_DEFAULT,
+    OsActivityStreamFlag,
+    OsTraceService,
+    SyslogEntry,
+    SyslogLogLevel,
+)
 from pymobiledevice3.services.syslog import SyslogService
 
 logger = logging.getLogger(__name__)
@@ -194,6 +200,8 @@ async def syslog_live(
     include_label: bool,
     regex: list[str],
     insensitive_regex: list[str],
+    no_debug: bool = False,
+    no_info: bool = False,
     image_offset: bool = False,
     start_after: Optional[str] = None,
     output_format: SyslogFormat = SyslogFormat.TEXT,
@@ -206,14 +214,27 @@ async def syslog_live(
         print(f'Waiting for "{start_after}" ...', flush=True)
 
     should_color_output = output_format is SyslogFormat.TEXT and user_requested_colored_output()
+    stream_flags = OS_TRACE_RELAY_STREAM_FLAGS_DEFAULT
+    if no_debug:
+        stream_flags &= ~OsActivityStreamFlag.DEBUG
+    if no_info:
+        stream_flags &= ~OsActivityStreamFlag.INFO
 
-    async for syslog_entry in OsTraceService(lockdown=service_provider).syslog(pid=pid):
+    async for syslog_entry in OsTraceService(lockdown=service_provider).syslog(pid=pid, stream_flags=stream_flags):
         if process_name and posixpath.basename(syslog_entry.filename) != process_name:
             continue
 
         if pid != -1 and syslog_entry.pid != pid:
             continue
 
+        if no_debug and syslog_entry.level == SyslogLogLevel.DEBUG:
+            continue
+
+        if no_info and syslog_entry.level == SyslogLogLevel.INFO:
+            # I don't really understand why INFO is retrieved when not requested, but this is imitating
+            # Console.app behavior
+            continue
+
         if output_format is SyslogFormat.JSON:
             json_line = format_json_line(syslog_entry)
             _emit_line(json_line, out)
@@ -350,6 +371,20 @@ async def cli_syslog_live(
             help="Include image offset in log line (text mode only; JSON always emits image_offset).",
         ),
     ] = False,
+    no_debug: Annotated[
+        bool,
+        typer.Option(
+            "--no-debug",
+            help="Suppress DEBUG entries.",
+        ),
+    ] = False,
+    no_info: Annotated[
+        bool,
+        typer.Option(
+            "--no-info",
+            help="Suppress INFO entries.",
+        ),
+    ] = False,
     start_after: Annotated[
         Optional[str],
         typer.Option(
@@ -382,6 +417,8 @@ async def cli_syslog_live(
             include_label,
             regex or [],
             insensitive_regex or [],
+            no_debug,
+            no_info,
             image_offset,
             start_after,
             output_format,

pymobiledevice3/services/os_trace.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python3
 import dataclasses
 import plistlib
 import struct
@@ -20,6 +19,54 @@
 SYSLOG_LINE_SPLITTER = "\n\x00"
 
 
+class OsActivityStreamFlag(IntEnum):
+    """
+    Exact `os_activity_stream_flag_t` values from Apple private headers.
+
+    Based on: https://github.com/limneos/oslog/blob/master/ActivityStreamAPI.h
+    """
+
+    PROCESS_ONLY = 0x00000001
+    SKIP_DECODE = 0x00000002
+    PAYLOAD = 0x00000004
+    HISTORICAL = 0x00000008
+    CALLSTACK = 0x00000010
+    DEBUG = 0x00000020
+    NO_SENSITIVE = 0x00000080
+    INFO = 0x00000100
+    PROMISCUOUS = 0x00000200
+
+
+class OsActivityStreamType(IntEnum):
+    """
+    Exact `os_activity_stream_type_t` values from Apple private headers.
+
+    Based on: https://github.com/limneos/oslog/blob/master/ActivityStreamAPI.h
+    """
+
+    ACTIVITY_CREATE = 0x0201
+    ACTIVITY_TRANSITION = 0x0202
+    ACTIVITY_USERACTION = 0x0203
+    TRACE_MESSAGE = 0x0300
+    LOG_MESSAGE = 0x0400
+    LEGACY_LOG_MESSAGE = 0x0480
+    TIMESYNC = 0x0500
+    SIGNPOST = 0x0600
+    LOSS = 0x0700
+    STATEDUMP_EVENT = 0x0A00
+
+
+OS_TRACE_RELAY_MESSAGE_FILTER_ALL = 0xFFFF
+
+# Sniffed from Console.app when enabling both info and debug messages
+OS_TRACE_RELAY_STREAM_FLAGS_DEFAULT = (
+    OsActivityStreamFlag.PAYLOAD
+    | OsActivityStreamFlag.HISTORICAL
+    | OsActivityStreamFlag.CALLSTACK
+    | OsActivityStreamFlag.DEBUG
+)
+
+
 class SyslogLogLevel(IntEnum):
     NOTICE = 0x00
     INFO = 0x01
@@ -229,14 +276,19 @@ async def collect(
                 await self.create_archive(f, size_limit=size_limit, age_limit=age_limit, start_time=start_time)
             TarFile(file).extractall(out)
 
-    async def syslog(self, pid=-1) -> typing.AsyncGenerator[SyslogEntry, None]:
+    async def syslog(
+        self,
+        pid: int = -1,
+        message_filter: int = OS_TRACE_RELAY_MESSAGE_FILTER_ALL,
+        stream_flags: int = OS_TRACE_RELAY_STREAM_FLAGS_DEFAULT,
+    ) -> typing.AsyncGenerator[SyslogEntry, None]:
         await self.connect()
         assert self.service is not None
         await self.service.send_plist({
             "Request": "StartActivity",
-            "MessageFilter": 65535,
+            "MessageFilter": message_filter,
             "Pid": pid,
-            "StreamFlags": 60,
+            "StreamFlags": stream_flags,
         })
 
         (length_length,) = struct.unpack("<I", await self.service.recvall(4))

tests/cli/test_syslog.py

@@ -6,18 +6,29 @@
 import pytest
 
 from pymobiledevice3.cli import syslog as syslog_module
-from pymobiledevice3.services.os_trace import SyslogEntry, SyslogLabel, SyslogLogLevel
+from pymobiledevice3.services.os_trace import (
+    OS_TRACE_RELAY_STREAM_FLAGS_DEFAULT,
+    OsActivityStreamFlag,
+    SyslogEntry,
+    SyslogLabel,
+    SyslogLogLevel,
+)
 
 pytestmark = [pytest.mark.cli]
 
 _FAKE_SERVICE_PROVIDER = object()
 
 
 class _FakeOsTraceService:
+    last_pid = None
+    last_stream_flags = None
+
     def __init__(self, entries):
         self._entries = entries
 
-    async def syslog(self, pid=-1):
+    async def syslog(self, pid=-1, stream_flags=OS_TRACE_RELAY_STREAM_FLAGS_DEFAULT, **_kwargs):
+        type(self).last_pid = pid
+        type(self).last_stream_flags = stream_flags
         for entry in self._entries:
             yield entry
 
@@ -34,6 +45,12 @@ def _create_syslog_entry(message: str) -> SyslogEntry:
     )
 
 
+def _create_syslog_entry_with_level(message: str, level: SyslogLogLevel) -> SyslogEntry:
+    entry = _create_syslog_entry(message)
+    entry.level = level
+    return entry
+
+
 def _create_syslog_entries(messages: list[str]) -> list[SyslogEntry]:
     return [_create_syslog_entry(message) for message in messages]
 
@@ -55,6 +72,8 @@ async def _run_syslog_live(
         "include_label": False,
         "regex": [],
         "insensitive_regex": [],
+        "no_debug": False,
+        "no_info": False,
     }
     kwargs.update(syslog_live_kwargs)
 
@@ -244,6 +263,50 @@ async def test_syslog_live_regex_filters_and_writes_plain_output_to_out(monkeypa
     assert out_lines == printed_lines
 
 
+@pytest.mark.asyncio
+async def test_syslog_live_no_debug_suppresses_only_debug(monkeypatch, capsys):
+    entries = [
+        _create_syslog_entry_with_level("info line", SyslogLogLevel.INFO),
+        _create_syslog_entry_with_level("debug line", SyslogLogLevel.DEBUG),
+        _create_syslog_entry_with_level("notice line", SyslogLogLevel.NOTICE),
+        _create_syslog_entry_with_level("error line", SyslogLogLevel.ERROR),
+    ]
+
+    printed_lines, _ = await _run_syslog_live(
+        monkeypatch,
+        capsys,
+        entries,
+        no_debug=True,
+    )
+
+    assert len(printed_lines) == 3
+    assert printed_lines[0].endswith("info line")
+    assert printed_lines[1].endswith("notice line")
+    assert printed_lines[2].endswith("error line")
+    assert _FakeOsTraceService.last_stream_flags == (OS_TRACE_RELAY_STREAM_FLAGS_DEFAULT & ~OsActivityStreamFlag.DEBUG)
+
+
+@pytest.mark.asyncio
+async def test_syslog_live_no_info_suppresses_only_info(monkeypatch, capsys):
+    entries = [
+        _create_syslog_entry_with_level("info line", SyslogLogLevel.INFO),
+        _create_syslog_entry_with_level("debug line", SyslogLogLevel.DEBUG),
+        _create_syslog_entry_with_level("notice line", SyslogLogLevel.NOTICE),
+    ]
+
+    printed_lines, _ = await _run_syslog_live(
+        monkeypatch,
+        capsys,
+        entries,
+        no_info=True,
+    )
+
+    assert len(printed_lines) == 2
+    assert printed_lines[0].endswith("debug line")
+    assert printed_lines[1].endswith("notice line")
+    assert _FakeOsTraceService.last_stream_flags == (OS_TRACE_RELAY_STREAM_FLAGS_DEFAULT & ~OsActivityStreamFlag.INFO)
+
+
 def test_format_json_line_with_label():
     entry = SyslogEntry(
         pid=42,