diff --git a/UDM_SEARCH_IMPLEMENTATION.md b/UDM_SEARCH_IMPLEMENTATION.md new file mode 100644 index 0000000..2399365 --- /dev/null +++ b/UDM_SEARCH_IMPLEMENTATION.md @@ -0,0 +1,158 @@ +# UDM Search Integration for Logstory + +## Overview + +Implemented **Option 1: Extend `_get_log_content()` to support UDM Search queries**. + +This feature allows logstory to fetch UDM events directly from Chronicle via UDM search queries, then replay them with updated timestamps into the ingestion pipeline—without needing pre-existing `.log` files. + +## Changes Made + +### 1. Core Logic (`src/logstory/main.py`) + +#### New Function: `_search_udm()` +- Executes UDM search queries against Chronicle API +- Parameters: + - `http_client`: Authenticated HTTP session + - `customer_id`: Chronicle customer ID + - `query`: UDM search query string + - `start_time`, `end_time`: Time range for search (defaults to last 30 days) + - `region`: Chronicle region (US or EU) +- Returns results as JSON lines (one UDM event per line) +- Handles API errors and logging + +#### Modified Function: `_get_log_content()` +- Now accepts optional UDM search parameters: + - `udm_query`: UDM search query to execute + - `http_client`: Authenticated client for the query + - `customer_id`: For API calls + - `region`: Chronicle region +- Logic flow: + - If `udm_query` provided → execute UDM search and return results + - Otherwise → fall back to existing file-based logic (GCS or local filesystem) +- Maintains full backward compatibility + +#### Modified Function: `usecase_replay_logtype()` +- Added parameters to support UDM queries +- Passes UDM parameters down to `_get_log_content()` +- Works seamlessly with existing timestamp replacement and ingestion pipeline + +### 2. CLI Commands (`src/logstory/logstory.py`) + +#### New Command: `replay from-udm-search` +```bash +logstory replay from-udm-search "" \ + --customer-id= \ + --credentials-path= \ + --timestamp-delta=1d +``` + +Example: +```bash +logstory replay from-udm-search "metadata.event_type='PROCESS_EXECUTION'" \ + --customer-id=01234567-0123-4321-abcd-01234567890a \ + --credentials-path=/path/to/credentials.json +``` + +#### New Internal Function: `_replay_from_udm()` +- Orchestrates UDM query execution and replay +- Applies timestamp shifting via existing pipeline +- Injects ingestion labels identifying the source as UDM search +- Outputs verification query for finding replayed events in Chronicle + +### 3. Configuration (`src/logstory/logtypes_events_timestamps.yaml`) + +Added timestamp configuration for UDM events: +```yaml +UDM_EVENTS: + api: udmevents + timestamps: + - name: created_time + base_time: true + pattern: '("created_time":\s*"?)(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(.\d+Z?\s*"?)' + dateformat: "%Y-%m-%dT%H:%M:%S" + group: 2 + - name: generic_event_timestamp + pattern: '("(?:event_time|timestamp|created_at|time)":\s*"?)(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(.\d+Z?\s*"?)' + dateformat: "%Y-%m-%dT%H:%M:%S" + group: 2 +``` + +- **api**: `udmevents` (uses native UDM ingestion) +- **timestamps**: Pattern matching for common timestamp field names in UDM events +- Supports ISO 8601 format timestamps (RFC 3339) + +### 4. Tests (`tests/test_udm_search.py`) + +Created comprehensive test suite: +- `test_search_udm_formats_results_as_jsonl()` - Verifies UDM API results conversion +- `test_get_log_content_with_udm_query()` - Tests UDM query path in content retrieval +- `test_get_log_content_without_udm_query()` - Ensures backward compatibility +- `test_udm_query_requires_http_client()` - Validates error handling + +## Data Flow + +``` +UDM Search Query + ↓ +_search_udm() → Chronicle API + ↓ +JSON Lines (one UDM event per line) + ↓ +usecase_replay_logtype() + ↓ +Timestamp extraction & replacement (via existing regex patterns) + ↓ +post_entries() → Ingestion API (udmevents) + ↓ +Chronicle with updated timestamps + ↓ +Ingestion Labels: + - log_replay: true + - replayed_from: logstory + - source_type: udm_search +``` + +## Key Design Decisions + +1. **No new file dependencies**: UDM results flow directly through logstory's existing replay pipeline +2. **Backward compatible**: Existing file-based replay unchanged; UDM is additive +3. **Timestamp handling**: Reuses pattern matching from YAML configs; ISO 8601 patterns added +4. **API abstraction**: Uses authenticated `http_client` from existing auth system +5. **User-facing**: CLI command is simple and discoverable + +## Usage Examples + +### Basic UDM search replay +```bash +logstory replay from-udm-search "metadata.event_type='PROCESS_EXECUTION'" +``` + +### With custom timestamp shift +```bash +logstory replay from-udm-search "event.principal.ip_address='192.168.1.1'" \ + --timestamp-delta=7d +``` + +### With environment variables +```bash +export LOGSTORY_CUSTOMER_ID= +export LOGSTORY_CREDENTIALS_PATH=/path/to/credentials.json +logstory replay from-udm-search "metadata.ingestion_time >= '2024-01-01'" +``` + +## Verification + +To find replayed logs: +``` +metadata.ingested_timestamp.seconds >= +metadata.ingestion_labels["source_type"]="udm_search" +``` + +## Future Enhancements + +- Support for UDM Search time range parameters in CLI +- Batch processing for large result sets +- Result pagination support +- Custom timestamp field mapping per query +- CSV export option (phase 2 of original issue) diff --git a/UDM_SEARCH_QUICKSTART.md b/UDM_SEARCH_QUICKSTART.md new file mode 100644 index 0000000..77ea233 --- /dev/null +++ b/UDM_SEARCH_QUICKSTART.md @@ -0,0 +1,106 @@ +# UDM Search Quick Start + +## What's New? + +Logstory now supports replaying logs directly from UDM searches without requiring pre-existing `.log` files. This is **Option 1** from issue #27: extend logstory's ingestion pipeline to accept UDM search results and apply timestamp shifting. + +## Quick Examples + +### Search for process execution events and replay with 1-day shift +```bash +logstory replay from-udm-search "metadata.event_type='PROCESS_EXECUTION'" \ + --customer-id=YOUR_CUSTOMER_ID \ + --credentials-path=/path/to/credentials.json +``` + +### Search for network connections from a specific IP +```bash +logstory replay from-udm-search "event.principal.ip_address='192.168.1.100'" \ + --timestamp-delta=7d +``` + +### Search with time filtering +```bash +logstory replay from-udm-search "metadata.event_type='USER_LOGIN' AND metadata.severity='WARNING'" +``` + +### Use environment variables for credentials +```bash +export LOGSTORY_CUSTOMER_ID=01234567-0123-4321-abcd-01234567890a +export LOGSTORY_CREDENTIALS_PATH=/path/to/credentials.json +export LOGSTORY_REGION=US + +logstory replay from-udm-search "metadata.ingestion_time >= '2024-01-01'" +``` + +## How It Works + +1. **Query UDM** — Logstory queries Chronicle's UDM Search API with your query +2. **Get Results** — Chronicle returns matching UDM events as JSON +3. **Shift Timestamps** — Logstory applies your timestamp delta (e.g., `--timestamp-delta=1d`) +4. **Ingest** — Updated events are replayed back into Chronicle +5. **Verify** — Special ingestion labels mark events as replayed from UDM search + +## Finding Your Replayed Logs + +After replay, search Chronicle with: +``` +metadata.ingested_timestamp.seconds >= +metadata.ingestion_labels["source_type"]="udm_search" +``` + +## Command Options + +``` +logstory replay from-udm-search [OPTIONS] QUERY + +Options: + --customer-id TEXT Chronicle customer ID + --credentials-path TEXT Path to service account JSON + --region TEXT Chronicle region (US or EU) + --timestamp-delta TEXT Time shift (e.g., 1d, 2h, 30m) + --env-file TEXT Path to .env file + --api-type TEXT API type (rest or legacy) + --project-id TEXT GCP project ID for REST API + --forwarder-name TEXT Forwarder name for REST API + --impersonate-service-account Service account to impersonate + --local-file-output Write to local files instead of API +``` + +## Use Cases + +- **Incident Response** — Replay security events with current timestamps to test detections +- **Testing** — Validate Chronicle configuration changes with historical UDM data +- **Demos** — Show security scenarios with realistic event sequences +- **Research** — Analyze subsets of historical data in a controlled replay scenario + +## What Changed in the Code? + +See `UDM_SEARCH_IMPLEMENTATION.md` for full technical details. + +### Files Modified: +- `src/logstory/main.py` — Core UDM search and replay logic +- `src/logstory/logstory.py` — New CLI command +- `src/logstory/logtypes_events_timestamps.yaml` — UDM event timestamp patterns +- `tests/test_udm_search.py` — Unit tests + +### Key New Functions: +- `_search_udm()` — Execute UDM search queries +- `_replay_from_udm()` — Orchestrate UDM search + replay + +## FAQ + +**Q: Do I still need `.log` files?** +A: No, not for UDM search replay. Regular file-based replay still works as before. + +**Q: Can I combine multiple UDM searches?** +A: Use a complex query: `(metadata.event_type='PROCESS_EXECUTION' OR metadata.event_type='FILE_MODIFICATION')` + +**Q: What timestamp delta values are supported?** +A: Combinations like `1d`, `2h`, `30m`, or `1d2h30m` + +**Q: How far back can I search?** +A: By default, the last 30 days. Customize with start/end times (future enhancement). + +**Q: Are replayed events marked differently?** +A: Yes, they have ingestion labels: `replayed_from=logstory` and `source_type=udm_search` diff --git a/src/logstory/logstory.py b/src/logstory/logstory.py index f5f365f..fd9da09 100644 --- a/src/logstory/logstory.py +++ b/src/logstory/logstory.py @@ -1040,6 +1040,52 @@ def replay_usecase_logtype( _replay_usecases(usecases, logtype_list, entities, timestamp_delta, local_file_output) +@replay_app.command("from-udm-search") +def replay_from_udm_search( + query: str = typer.Argument(..., help="UDM search query"), + env_file: str | None = EnvFileOption, + credentials_path: str | None = CredentialsOption, + customer_id: str | None = CustomerIdOption, + region: str | None = RegionOption, + timestamp_delta: str | None = TimestampDeltaOption, + local_file_output: bool = LocalFileOutputOption, + api_type: str | None = ApiTypeOption, + project_id: str | None = ProjectIdOption, + forwarder_name: str | None = ForwarderNameOption, + impersonate_service_account: str | None = ImpersonateServiceAccountOption, +): + """Replay logs from a UDM search query. + + Args: + query: UDM search query (e.g., "metadata.event_type='PROCESS_EXECUTION'") + """ + final_credentials, final_customer_id, final_region = _load_and_validate_params( + env_file, + credentials_path, + customer_id, + region, + impersonate_service_account, + api_type, + ) + _set_environment_vars( + final_credentials, + final_customer_id, + final_region, + api_type, + project_id, + forwarder_name, + impersonate_service_account, + ) + + _replay_from_udm( + query, + timestamp_delta, + final_customer_id, + final_region, + local_file_output, + ) + + def _replay_usecases( usecases: list[str], logtypes: list[str] | str, @@ -1083,6 +1129,61 @@ def _replay_usecases( """) +def _replay_from_udm( + query: str, + timestamp_delta: str | None, + customer_id: str, + region: str, + local_file_output: bool = False, +): + """Replay logs from UDM search query. + + Args: + query: UDM search query. + timestamp_delta: Timestamp delta for time shifts. + customer_id: Chronicle customer ID. + region: Chronicle region. + local_file_output: Whether to write to local files instead of API. + """ + logstory_exe_time = _get_current_time() + + typer.echo(f"Executing UDM search query: {query}") + + # Get authenticated HTTP client + if not imported_main.http_client: + raise RuntimeError("No HTTP client available for UDM search") + + # We need a synthetic log_type and use_case for the replay pipeline + # Since this is a direct UDM search, we'll use generic names + use_case = "UDM_SEARCH" + log_type = "UDM_EVENTS" + + try: + old_base_time = imported_main.usecase_replay_logtype( + use_case, + log_type, + logstory_exe_time, + timestamp_delta=timestamp_delta, + entities=False, + local_file_output=local_file_output, + udm_query=query, + http_client=imported_main.http_client, + customer_id=customer_id, + region=region, + ) + + typer.echo(f"Successfully replayed UDM search results with timestamp delta: {timestamp_delta}") + typer.echo(f"""UDM Search for the replayed logs: + metadata.ingested_timestamp.seconds >= {int(logstory_exe_time.timestamp())} + metadata.ingestion_labels["log_replay"]="true" + metadata.ingestion_labels["replayed_from"]="logstory" + metadata.ingestion_labels["source_type"]="udm_search" + """) + except Exception as e: + typer.echo(f"Error replaying UDM search: {e}", err=True) + raise typer.Exit(code=1) + + def entry_point(): """Main entry point for the CLI.""" app() diff --git a/src/logstory/logtypes_events_timestamps.yaml b/src/logstory/logtypes_events_timestamps.yaml index e2e7408..4fe0e7e 100644 --- a/src/logstory/logtypes_events_timestamps.yaml +++ b/src/logstory/logtypes_events_timestamps.yaml @@ -748,3 +748,16 @@ GITHUB: pattern: '("@timestamp":)(\d{10})' dateformat: 'epoch' group: 2 + +UDM_EVENTS: + api: udmevents + timestamps: + - name: created_time + base_time: true + pattern: '("created_time":\s*"?)(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(.\d+Z?\s*"?)' + dateformat: "%Y-%m-%dT%H:%M:%S" + group: 2 + - name: generic_event_timestamp + pattern: '("(?:event_time|timestamp|created_at|time)":\s*"?)(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(.\d+Z?\s*"?)' + dateformat: "%Y-%m-%dT%H:%M:%S" + group: 2 diff --git a/src/logstory/main.py b/src/logstory/main.py index ddc5aa5..3d1288d 100644 --- a/src/logstory/main.py +++ b/src/logstory/main.py @@ -314,10 +314,92 @@ def _validate_timestamp_config(log_type: str, timestamp_map: dict[str, Any]) -> LOGGER.debug("Timestamp configuration validation passed for log type '%s'", log_type) +def _search_udm( + http_client: requests.AuthorizedSession, + customer_id: str, + query: str, + start_time: datetime.datetime | None = None, + end_time: datetime.datetime | None = None, + region: str = "US", +) -> str: + """Searches UDM events and returns results as JSON lines. + + Args: + http_client: Authenticated HTTP client for API calls. + customer_id: Chronicle customer ID. + query: UDM search query. + start_time: Start time for search (defaults to 30 days ago). + end_time: End time for search (defaults to now). + region: Chronicle region (US or EU). + + Returns: + JSON lines string with UDM event results. + """ + url_prefix = f"{region.lower()}-" if region.upper() != "US" else "" + api_url = ( + f"https://{url_prefix}malachiteingestion-pa.googleapis.com/" + f"v1/projects/{customer_id}:search" + ) + + if end_time is None: + end_time = datetime.datetime.now(datetime.UTC) + if start_time is None: + start_time = end_time - datetime.timedelta(days=30) + + payload = { + "query": query, + "start_time": start_time.isoformat(), + "end_time": end_time.isoformat(), + } + + LOGGER.info("Executing UDM search query: %s", query) + try: + response = http_client.post(api_url, json=payload) + response.raise_for_status() + + results = [] + for line in response.text.strip().split('\n'): + if line: + event_data = json.loads(line) + results.append(json.dumps(event_data)) + + LOGGER.info("UDM search returned %d results", len(results)) + return '\n'.join(results) + except Exception as e: + LOGGER.error("Error searching UDM: %s", e) + raise + + def _get_log_content( - use_case: str, log_type: str, entities: bool | None = False + use_case: str, + log_type: str, + entities: bool | None = False, + udm_query: str | None = None, + http_client: requests.AuthorizedSession | None = None, + customer_id: str | None = None, + region: str = "US", ) -> str: - """Retrieves log content from either GCS or local filesystem.""" + """Retrieves log content from GCS, local filesystem, or UDM search. + + Args: + use_case: Use case name for file paths. + log_type: Log type name for file paths. + entities: Whether to retrieve entities instead of events. + udm_query: Optional UDM search query to execute instead of reading files. + http_client: Authenticated HTTP client for UDM search. + customer_id: Chronicle customer ID for UDM search. + region: Chronicle region for UDM search. + + Returns: + Log content as string. + """ + if udm_query: + if not http_client or not customer_id: + raise ValueError( + "http_client and customer_id required when using udm_query" + ) + return _search_udm(http_client, customer_id, udm_query, region=region) + if entities: object_name = f"{use_case}/ENTITIES/{log_type}.log" else: @@ -644,6 +726,10 @@ def usecase_replay_logtype( ts_map_path: str | None = "./", entities: bool | None = False, local_file_output: bool = False, + udm_query: str | None = None, + http_client: requests.AuthorizedSession | None = None, + customer_id: str | None = None, + region: str | None = None, ) -> datetime.datetime | None: """Replays log data for a specific use case and log type. @@ -657,6 +743,10 @@ def usecase_replay_logtype( ts_map_path: disk location of the yaml files entities: bool for Entities (True) vs Events (False) local_file_output: bool to write to local files instead of API + udm_query: Optional UDM search query to execute instead of reading files. + http_client: Authenticated HTTP client for UDM search. + customer_id: Chronicle customer ID for UDM search. + region: Chronicle region for UDM search. Returns: old_base_time: so that subsequent logtypes/usecases can all use the same value @@ -681,7 +771,15 @@ def usecase_replay_logtype( api_for_log_type = timestamp_map[log_type]["api"] # Get optional log_dir from YAML config, defaults to None for backwards compatibility log_type_log_dir = timestamp_map[log_type].get("log_dir") - log_content = _get_log_content(use_case, log_type, entities) + log_content = _get_log_content( + use_case, + log_type, + entities, + udm_query=udm_query, + http_client=http_client, + customer_id=customer_id, + region=region or REGION, + ) ingestion_labels = _get_ingestion_labels( use_case, logstory_exe_time, api_for_log_type ) diff --git a/tests/test_udm_search.py b/tests/test_udm_search.py new file mode 100644 index 0000000..1c7c961 --- /dev/null +++ b/tests/test_udm_search.py @@ -0,0 +1,113 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Tests for UDM search integration.""" + +import json +from unittest import mock + +import pytest + +# Import the main module functions +import sys +sys.path.insert(0, '../src') +from logstory import main + + +def test_search_udm_formats_results_as_jsonl(): + """Test that _search_udm converts API results to JSON lines format.""" + mock_http_client = mock.MagicMock() + + # Mock the API response with JSON lines format (one per line) + udm_results = [ + {"event": {"created_time": "2024-01-01T10:00:00Z"}, "metadata": {"event_id": "1"}}, + {"event": {"created_time": "2024-01-01T10:01:00Z"}, "metadata": {"event_id": "2"}}, + ] + response_text = '\n'.join(json.dumps(r) for r in udm_results) + mock_http_client.post.return_value.text = response_text + mock_http_client.post.return_value.raise_for_status = mock.MagicMock() + + result = main._search_udm( + mock_http_client, + "test-customer-id", + "metadata.event_type='PROCESS_EXECUTION'", + region="US" + ) + + # Verify result is JSON lines format + lines = result.strip().split('\n') + assert len(lines) == 2 + + # Verify each line is valid JSON + for line in lines: + event = json.loads(line) + assert "event" in event + assert "metadata" in event + + +def test_get_log_content_with_udm_query(): + """Test that _get_log_content works with UDM queries.""" + mock_http_client = mock.MagicMock() + + udm_result = '{"event": {"created_time": "2024-01-01T10:00:00Z"}}' + mock_http_client.post.return_value.text = udm_result + mock_http_client.post.return_value.raise_for_status = mock.MagicMock() + + with mock.patch.object(main, '_search_udm') as mock_search: + mock_search.return_value = udm_result + + result = main._get_log_content( + "UDM_SEARCH", + "UDM_EVENTS", + entities=False, + udm_query="metadata.event_type='PROCESS_EXECUTION'", + http_client=mock_http_client, + customer_id="test-customer-id", + region="US" + ) + + assert result == udm_result + mock_search.assert_called_once() + + +def test_get_log_content_without_udm_query(): + """Test that _get_log_content still works for file-based logs.""" + # Mock the storage client to return None (local file case) + with mock.patch.object(main, 'storage_client', None): + with mock.patch('builtins.open', mock.mock_open(read_data='test log content')): + result = main._get_log_content( + "TEST_USECASE", + "TEST_LOGTYPE", + entities=False + ) + + assert result == 'test log content' + + +def test_udm_query_requires_http_client(): + """Test that UDM query without HTTP client raises error.""" + with pytest.raises(ValueError) as exc_info: + main._get_log_content( + "UDM_SEARCH", + "UDM_EVENTS", + entities=False, + udm_query="metadata.event_type='PROCESS_EXECUTION'", + http_client=None, + customer_id="test-customer-id" + ) + + assert "http_client and customer_id required" in str(exc_info.value) + + +if __name__ == '__main__': + pytest.main([__file__, '-v'])