diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..6fff16c --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,6 @@ +version: 2 +updates: + - package-ecosystem: github-actions + directory: "/" + schedule: + interval: monthly diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 698e959..fccdbdc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,6 +4,9 @@ on: push: pull_request: +permissions: + contents: read + jobs: test: runs-on: ubuntu-latest @@ -12,19 +15,36 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.11", "3.12", "3.13"] + python-version: ["3.11", "3.12", "3.13", "3.14"] steps: - - uses: actions/checkout@v4 - - uses: actions/setup-python@v5 + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 with: python-version: ${{ matrix.python-version }} + - name: Install pinned build tooling + run: python -m pip install --disable-pip-version-check build==1.3.0 twine==7.0.0 setuptools==84.0.0 - name: Run unit tests run: python -m unittest discover -s tests -v - name: Compile source run: python -m compileall -q src tests - name: Build wheel and sdist - run: python -m pip wheel --no-deps --wheel-dir dist . - - name: Install and smoke-test wheel + env: + SOURCE_DATE_EPOCH: 0 + run: python -m build --sdist --wheel --outdir dist + - name: Validate metadata and assets + run: | + python -m twine check --strict dist/* + test "$(find dist -name '*.whl' | wc -l)" -eq 1 + test "$(find dist -name '*.tar.gz' | wc -l)" -eq 1 + - name: Fresh-install wheel smoke test + run: | + env -u PYTHONPATH python -m venv "$RUNNER_TEMP/wheel-env" + env -u PYTHONPATH "$RUNNER_TEMP/wheel-env/bin/python" -m pip install --force-reinstall --no-deps dist/*.whl + env -u PYTHONPATH "$RUNNER_TEMP/wheel-env/bin/monium" --version + - name: Fresh-install sdist smoke test run: | - python -m pip install --force-reinstall --no-deps dist/*.whl - env -u PYTHONPATH monium --version + env -u PYTHONPATH python -m venv "$RUNNER_TEMP/sdist-env" + env -u PYTHONPATH "$RUNNER_TEMP/sdist-env/bin/python" -m pip install --force-reinstall --no-deps dist/*.tar.gz + env -u PYTHONPATH "$RUNNER_TEMP/sdist-env/bin/monium" --version diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..6d3a2f2 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,100 @@ +name: Release + +on: + push: + tags: + - "v*" + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + env: + PYTHONPATH: src + strategy: + fail-fast: false + matrix: + python-version: ["3.11", "3.12", "3.13", "3.14"] + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: ${{ matrix.python-version }} + - name: Test source + run: | + python -m unittest discover -s tests -v + python -m compileall -q src tests + + publish: + needs: test + if: github.repository == 'kizz-tech/monium' + runs-on: ubuntu-latest + permissions: + contents: write + id-token: write + attestations: write + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + fetch-depth: 0 + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.13" + - name: Validate annotated tag and package version + run: | + test "$(git cat-file -t "$GITHUB_REF_NAME")" = "tag" + tag_commit="$(git rev-list -n 1 "$GITHUB_REF_NAME")" + main_commit="$(git rev-parse refs/remotes/origin/main)" + test "$tag_commit" = "$main_commit" + tag_version="${GITHUB_REF_NAME#v}" + package_version="$(PYTHONPATH=src python -c 'from monium_cli import __version__; print(__version__)')" + test "$tag_version" = "$package_version" + - name: Install pinned build tooling + run: python -m pip install --disable-pip-version-check build==1.3.0 twine==7.0.0 setuptools==84.0.0 + - name: Set source epoch + run: echo "SOURCE_DATE_EPOCH=$(git log -1 --format=%ct)" >> "$GITHUB_ENV" + - name: Build wheel and sdist once + run: python -m build --sdist --wheel --outdir dist + - name: Validate metadata and asset count + run: | + python -m twine check --strict dist/* + test "$(find dist -name '*.whl' | wc -l)" -eq 1 + test "$(find dist -name '*.tar.gz' | wc -l)" -eq 1 + - name: Fresh-install release assets + run: | + python -m venv "$RUNNER_TEMP/wheel-env" + "$RUNNER_TEMP/wheel-env/bin/python" -m pip install --no-deps dist/*.whl + env -u PYTHONPATH "$RUNNER_TEMP/wheel-env/bin/monium" --version + python -m venv "$RUNNER_TEMP/sdist-env" + "$RUNNER_TEMP/sdist-env/bin/python" -m pip install --no-deps dist/*.tar.gz + env -u PYTHONPATH "$RUNNER_TEMP/sdist-env/bin/monium" --version + printf '{"message":"ready","service":"telemetry"}\n' | env -u PYTHONPATH "$RUNNER_TEMP/wheel-env/bin/monium" logs parse - --input-format ndjson --summary-only + - name: Generate checksums + run: | + cd dist + sha256sum *.whl *.tar.gz > SHA256SUMS + sha256sum --check SHA256SUMS + - name: Attest release assets + uses: actions/attest-build-provenance@4d101475d8b20a2381f78447822ac1eab6504dd8 # v4.2.2 + with: + subject-path: | + dist/*.whl + dist/*.tar.gz + dist/SHA256SUMS + - name: Publish GitHub release + env: + GH_TOKEN: ${{ github.token }} + run: | + case "$GITHUB_REF_NAME" in + *a*|*b*|*rc*) + gh release create "$GITHUB_REF_NAME" --verify-tag --generate-notes --title "Unofficial CLI for Yandex Monium $GITHUB_REF_NAME" --prerelease dist/*.whl dist/*.tar.gz dist/SHA256SUMS + ;; + *) + gh release create "$GITHUB_REF_NAME" --verify-tag --generate-notes --title "Unofficial CLI for Yandex Monium $GITHUB_REF_NAME" dist/*.whl dist/*.tar.gz dist/SHA256SUMS + ;; + esac diff --git a/.gitignore b/.gitignore index 2af1de2..dd0eab3 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ __pycache__/ *.py[cod] +.DS_Store *.egg-info/ .pytest_cache/ .mypy_cache/ diff --git a/CHANGELOG.md b/CHANGELOG.md index b33d888..5165656 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,10 +1,34 @@ # Changelog -All notable changes to Monium CLI are documented here. +## 0.2.0rc1 — 2026-08-11 -## [0.1.0] - 2026-08-11 +- Reframed the package as an explicitly offline, credential-free toolkit. +- Added `logs query`, typed repeated selector clauses, critical-path support, + strict trace literal semantics, and HTTPS-origin validation. +- Added bounded NDJSON/JSON/CSV/text readers with neutral and OpenTelemetry + normalization, AnyValue handling, provenance, bounded unknown-field + preservation, structured errors, and summary-only omission markers. Records + over the field cap fail closed; output is independently capped by bytes and + JSON nodes with explicit row-omission warnings and an exact final CLI-envelope + size check. +- Added optional non-secret TOML/environment configuration with deterministic + CLI precedence and effective-value echoing. +- Rebuilt packaging and CI for Python 3.11–3.14, single-build verified and + attested release assets, metadata validation, and fresh-install smoke tests. +- Breaking change: removed the legacy path/status selector shortcuts and their + Python keyword arguments; use generic typed field filters instead. +- Duration selectors accept `us`, `ms`, `s`, `m`, and `h` (not `ns`, `µs`, or + `d`). Critical-path selectors use `span.critical_path` PRESENT/ABSENT. +- Added `monium capabilities`, which reports offline support and explicitly + marks native live Alerts/Logs reads unavailable without probing credentials, + network, or a browser. Private console APIs are not packaged as providers. +- Documented the fail-closed live-read release gate. No speculative provider, + transport model, alert wire schema, or live plugin API is packaged in this + offline release candidate. +- Rejected generic filters that attempt to override first-class project, + cluster, service, log-group, trace, or span scope; sanitized config/export + read errors so credential-like path values are not echoed. -- Added offline `monium logs url` URL and selector builder. -- Added offline `monium traces query` selector builder. -- Added `monium logs parse` NDJSON normalization, filters, summaries, and clear malformed-input errors. -- Added standard-library unit tests, packaging metadata, and GitHub CI. +## 0.1.0 — 2026-08-11 + +- Initial offline selector builders and NDJSON parser. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index a023aaa..f1ffdd4 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -1,10 +1,13 @@ # Contributing -Thanks for helping improve Monium CLI. Keep changes small and focused. +Contributions to the current offline release line should keep the project +credential-free and based on Python's standard library at runtime. Do not add +uploads, background telemetry, private-project fixtures, private console +gateways, browser/session fallbacks, or custom live endpoints. A future live +provider is acceptable only after the official API contract and every release +gate documented in the README and SECURITY policy are available and tested. -## Local checks - -Use Python 3.11 or newer and run: +Before opening a pull request: ```console make test @@ -12,10 +15,29 @@ make check make build ``` -Runtime code must remain standard-library-only. Tests must be deterministic, -offline, and must not launch a browser or use credentials. Add or update unit -tests for behavior changes and keep JSON output stable for automation. +Tests are table-driven `unittest` cases and must not require network access or +credentials. New parsers should enforce input bounds, preserve source +provenance and unknown fields, and emit structured errors without tracebacks. +Selector changes should keep typed operators and reject ambiguous/raw +injection. Update the README and CHANGELOG when public behaviour changes. + +Use focused commits and include the exact validation commands in the pull +request description. Release publication is handled by maintainers through the +tag workflow; this project does not publish to PyPI. + +## Maintainer release checklist + +1. Confirm the working tree contains only the intended release changes and the + package version matches the planned `v` tag. +2. Run `make test`, `make check`, `make build`, strict Twine validation, and + fresh-install smoke tests for both wheel and sdist. +3. Push the commit through a pull request and wait for every Python CI matrix + job to succeed before merging. +4. Create an annotated, version-matching tag from the verified `main` commit, + for example `git tag -a v0.2.0rc1 -m "Monium CLI v0.2.0rc1"`. +5. Push only that tag, wait for the Release workflow, then verify the public + release state, asset count, `SHA256SUMS`, provenance, and a fresh install of + the downloaded wheel. -Please do not add Monium credentials, exported production data, or secrets to -issues, commits, fixtures, or logs. By submitting a contribution, you agree -that it may be distributed under the Apache-2.0 license in this repository. +Never move or reuse an existing release tag. Publish a corrective version and +revert through normal review if a release candidate is defective. diff --git a/MANIFEST.in b/MANIFEST.in new file mode 100644 index 0000000..0ba0866 --- /dev/null +++ b/MANIFEST.in @@ -0,0 +1,3 @@ +include CHANGELOG.md +include CONTRIBUTING.md +include SECURITY.md diff --git a/Makefile b/Makefile index acc7517..37972e8 100644 --- a/Makefile +++ b/Makefile @@ -7,4 +7,4 @@ check: PYTHONPATH=src python -m compileall -q src tests build: - python -m pip wheel --no-deps --wheel-dir dist . + python -m build --sdist --wheel --outdir dist diff --git a/README.md b/README.md index 34aba7f..b46c4f0 100644 --- a/README.md +++ b/README.md @@ -1,129 +1,207 @@ -# Monium CLI - -Monium CLI is a small, unofficial Python command-line tool for repeatable Monium Logs and Traces workflows. It builds links and selector queries locally and parses an exported NDJSON file locally. Version 0.1.0 makes no live API calls and has no authentication or telemetry. +# Unofficial CLI for Yandex Monium + +`monium` is a small, credential-free Python toolkit for building typed Logs +and Traces selectors and parsing common telemetry exports locally. Version +`0.2.0rc1` is explicitly offline: core commands perform no API calls, uploads, +metrics, background telemetry, authentication, or network access. The opt-in +`logs url --open` flag launches the system browser; any navigation/network +activity after that is controlled by the browser, not by the CLI. + +## Capability matrix + +| Area | Supported | Boundary | +| --- | --- | --- | +| Offline builders | Logs query/URL and Traces query selectors, typed operators, critical-path flag | Does not verify that a project or field exists | +| Offline parsers | Generic NDJSON, JSON, CSV, text adapters; neutral rows and documented OTLP JSON envelopes | JSON is materialized with a bounded input size; formats are not a claim of exact Monium export-schema compatibility | +| UI adapter | `logs url` emits a validated HTTPS-origin route | UI route is volatile and may change; `--open` is opt-in | +| Live/authenticated operations | Not registered | Current Yandex documentation does not publish a supported native Alerts/Logs read transport with enforceable project/cluster/service scope | + +`monium capabilities` reports this matrix as bounded JSON without reading +credentials, probing the network, or attempting a browser fallback. In +particular, it reports `alerts.list`, `alerts.read`, and `logs.query.live` as +unavailable with `reason_code: no_supported_provider`; the CLI does not +advertise placeholder commands that cannot complete safely. + +### Live-read release gate + +The unavailable entries are deliberate. As of 2026-08-11, the official +[Monium API catalog](https://yandex.cloud/en/docs/monium/api-ref/) publishes no +native Alerts list/read or Monium Logs read/query endpoint with a request +schema, response schema, pagination, and authentication contract. Roles and +[query-language documentation](https://yandex.cloud/en/docs/monium/logs/query) +do not fill that gap, and the documented +[legacy Cloud Logging reader](https://yandex.cloud/en/docs/logging/api-ref/grpc/LogReading/read) +does not implement native Monium project/cluster/service semantics. + +The three live capabilities remain unavailable until one supported provider +and official contract fixtures exist. The first implementation must satisfy +all of these conditions together: + +- model alert ownership as `AlertScope(project)` and telemetry reads as + `TelemetryScope(project, cluster, service)`; a response echo is correlation + data, never proof of server-side authorization or scope enforcement; +- decode the official alert definition as a closed discriminated union of + `threshold`, `expression`, and `multialert`, preserving multiple queries, + no-data variants, notifications, labels, and annotations; unknown semantics + fail closed instead of being guessed or silently dropped; +- report any output transformation with explicit `redacted_fields` and + `definition_complete`; heuristic redaction cannot claim a complete + definition; +- enforce total request byte and node limits before authentication or provider + invocation, then stream responses under hard byte caps, connect/read/total + deadlines, and cancellation; +- prove list-to-read alert ID round trips, cross-scope and reserved-filter + rejection, bounded pagination, oversized-response handling, and credential + non-disclosure in arguments, output, and errors; +- use only a fixed, documented official endpoint and auth contract. No private + console gateway, browser/session fallback, redirect-derived endpoint, or + custom live endpoint is permitted. + +Until those gates are evidenced, rollback and normal behavior are identical: +no live command is registered and the capability reason remains +`no_supported_provider`. + +The parser preserves unknown record fields under `fields` up to the documented +`MAX_FIELDS` cap (2,000 per record), normalized resource metadata under +`attributes`, and source line/byte provenance. Records that exceed the field +cap fail closed instead of silently dropping data. Successful parser payloads +are additionally bounded to 8 MiB and 1,000,000 JSON nodes; rows that do not fit +are omitted with an explicit warning and `rows_omitted: true`. It does not claim +exact compatibility with any private or UI export format beyond the generic +formats listed above. ## Install -Python 3.11 or newer is required. From a checkout: +Python 3.11–3.14 is supported. Install a release from GitHub or use `pipx`: ```console -python -m pip install . -monium --version +pipx install https://github.com/kizz-tech/monium/releases/download/v0.2.0rc1/monium_cli-0.2.0rc1-py3-none-any.whl ``` -For development, the package can also be used without installation: +From a checkout: ```console -PYTHONPATH=src python -m monium_cli --version +python -m pip install . +monium --version ``` -## Build a Logs link +No runtime dependency outside Python's standard library is required. + +## Build selectors -The project and service are required. Optional selectors are escaped and validated before they are placed in the query. `--open` is opt-in; without it, no browser is started. +Logs query requires `--project` plus `--service`, or accepts `--log-group-id` +alone or `--trace-id` alone. Repeated typed filters are available as `--field KEY=VALUE`, +`--glob-field KEY=PATTERN`, `--regex-field KEY=PATTERN`, +`--contains-field KEY=TEXT` (only `message` or `meta.*`), +`--number-field KEY=NUMBER` for numeric equality, and +`--number FIELD OP VALUE` (`=`, `!=`, `>`, `>=`, `<`, `<=`). There are no +domain-specific request-path or status-code shortcuts. ```console -monium logs url \ - --project folder__demo \ - --cluster production \ - --service payments \ - --level ERROR \ - --contains "payment failed" \ - --from now-2h \ - --to now -``` +monium logs query --project demo --service api \ + --contains "timeout" --field region=eu --number latency_ms ">=" 500 + +monium logs url --project demo --service api --from now-2h --to now -Example (formatted for readability): - -```json -{ - "command": "logs url", - "ok": true, - "open_requested": false, - "opened": false, - "project": "folder__demo", - "query": "{ project==\"folder__demo\", cluster==\"production\", service==\"payments\", level==ERROR, message=*\"payment failed\" }", - "url": "https://monium.yandex.cloud/projects/folder__demo/logs?query=...&from=now-2h&to=now&columns=level%2Ctime%2Cmessage%2Chost&tab=logs" -} +monium traces query --project demo --trace-id 4bf92f3577b34da6a3ce929d0e0e4736 \ + --span-id 123 --span-status ERROR --min-duration 1s --critical-path ``` -The URL path and `query/from/to/columns/tab` parameters follow the current Monium Logs UI shape. The CLI does not verify that a project exists. +Trace IDs, span IDs, statuses, names, and other strings remain quoted even if +they look like `123`, `true`, or `1s`; only explicitly typed durations and +numbers are unquoted. Supported duration units are `us`, `ms`, `s`, `m`, and +`h`. -## Build a Traces query +`logs url` validates an HTTPS origin and never opens a browser unless `--open` +is present. The URL route is a volatile UI adapter, not a live API contract. -```console -monium traces query \ - --project folder__demo \ - --service payments \ - --trace-id 4bf92f3577b34da6a3ce929d0e0e4736 \ - --operation "POST /charge" \ - --min-duration 1s -``` +## Parse exports -Output is deterministic JSON such as: +Use an explicit format for stdin; `auto` uses safe file extensions/content +recognition and rejects ambiguous stdin: -```json -{ - "command": "traces query", - "ok": true, - "project": "folder__demo", - "query": "{ project==\"folder__demo\", service==\"payments\", trace.id==\"4bf92f3577b34da6a3ce929d0e0e4736\", span.name==\"POST /charge\", span.duration>=1s }" -} +```console +monium logs parse export.ndjson --input-format ndjson --limit 50 +cat export.csv | monium logs parse - --input-format csv --summary-only +monium logs parse export.json --input-format json --include-attributes ``` -`--operation` maps to the Monium `span.name` field. Current Monium selector semantics are used: `==` for exact values, `=*` for substring matching, and comparison operators for durations. - -## Parse an exported NDJSON file - -Export the file from Monium, then parse it without credentials or network access: +Supported formats stream line-by-line where practical. JSON input is +materialized and capped; line, total-input, nesting, integer, and summary +cardinality bounds produce structured JSON errors. Output has independent hard +byte and node budgets; budgeted row omission is always reported rather than +silently truncating fields. The CLI remeasures the complete formatted envelope, +including its final newline, and fails closed with a small non-zero JSON error +if envelope formatting itself would exceed either cap. `--summary-only` sets +`rows_omitted: true`; +`truncated` still reports whether the row limit would have omitted matching +rows. + +The CLI is single-threaded. Python's CSV field-size setting is process-global, +so `parse_export(..., input_format="csv")` serializes its own CSV calls; code +that changes `csv.field_size_limit()` concurrently from outside this package is +not a supported library usage. + +### 0.2 migration note + +The 0.2 selector API intentionally removes the legacy path/status shortcuts +and their corresponding Python keyword arguments. The old path shortcut was a +contains operation on a schema-specific metadata field; the old status +shortcut was numeric exact equality on another schema-specific field. Choose +your own exported schema keys with generic typed replacements, for example +`--contains-field meta.route=/ready` and `--number-field telemetry.status=503`. +OpenTelemetry names such as `http.route`/`http.status_code` are schema- +dependent alternatives, not guaranteed equivalents. This is an explicit +breaking change within the pre-1.0 API. + +## Configuration + +Configuration is optional and non-secret. A TOML file is read only when named +with `--config`; environment values then override it, and explicit CLI values +override environment values. Only the allowlisted `MONIUM_*` names matching +these keys are read; other environment variables are ignored. Allowed keys are `project`, `service`, `cluster`, +`ui_base_url`, `from_time`, `to_time`, `columns`, and `input_format`. + +```toml +project = "demo" +service = "api" +cluster = "production" +ui_base_url = "https://monium.yandex.cloud" +columns = ["level", "time", "message"] +``` ```console -monium logs parse export.ndjson \ - --service payments \ - --level ERROR WARN \ - --contains timeout \ - --limit 50 +MONIUM_SERVICE=worker monium --config monium.toml logs query +monium --no-config logs query --project demo --service api ``` -The parser accepts flat Monium rows and common OpenTelemetry forms (camelCase or snake_case fields, `Resource.attributes`, `scopeLogs/logRecords`, and `AnyValue` wrappers). It returns a summary and compact rows. `--include-attributes` includes normalized resource/meta/label attributes. - -```json -{ - "command": "logs parse", - "file": "export.ndjson", - "matched": 1, - "ok": true, - "returned": 1, - "rows": [ - { - "cluster": "production", - "level": "ERROR", - "line": 7, - "message": "payment timeout", - "operation": null, - "request_path": "/charge", - "service": "payments", - "span_id": "e7a886", - "status_code": 504, - "time": "2026-08-11T10:00:00Z", - "trace_id": "4bf92f..." - } - ], - "summary": {"levels": {"ERROR": 1}, "services": {"payments": 1}}, - "total_scanned": 7, - "truncated": false -} -``` +Unknown TOML keys—including credential-like key names—are rejected. Values are +not secret-scanned, so do not place secrets in configuration. No token, cookie, +or credential environment variable is handled; only the allowlisted names +above are read. Successful responses echo the effective configured values in a +`config` object. + +## Security and privacy -Malformed JSON is a hard error with a line number and a non-zero exit code; partial output is not presented as complete. All normal command and error responses are JSON, which makes the CLI suitable for scripts. +The CLI is offline and credential-free. It does not read tokens/cookies or +send files anywhere. Avoid placing secrets in command arguments, shell +history, exports, config files, or bug reports. Malformed input is reported as +bounded JSON errors without tracebacks. See [SECURITY.md](SECURITY.md) for +reporting guidance. -## Limitations and privacy +The package contains no adapter for the Monium web console's private gateway. +It never imports a browser session, cookies, CSRF state, or private console +routes. A future live provider must use a supported upstream contract and must +prove its server-side scope semantics before its commands can become +available. Client-side response echoes or filtering do not constitute that +proof. -- v0.1.0 has no Monium API client, Yandex Cloud SDK, authentication flow, retries, uploads, or background telemetry. -- URL generation assumes the Monium web UI route shown above; the UI may change independently of this package. -- NDJSON exports are read locally as UTF-8. Unrecognized fields are ignored unless they are normalized attributes. -- Selector values are bounded, validated, and escaped. Do not put credentials or other secrets in command arguments, exported files, shell history, or bug reports. +For service documentation, consult the [official Monium +documentation](https://yandex.cloud/docs/monium/). -This project is not affiliated with, sponsored by, or endorsed by Monium, Yandex Cloud, or their respective trademark owners. “Monium” and “Yandex Cloud” are used only to identify the service this unofficial tool can interoperate with. +This project is not affiliated with, sponsored by, or endorsed by Yandex +Cloud. “Monium” and “Yandex Cloud” are used only to identify the service with +which this unofficial tool interoperates. ## Development @@ -133,4 +211,6 @@ make check make build ``` -The project uses only Python's standard library at runtime. Tests never access the network or launch a browser. +Tests use only the standard library and never require network access or +credentials. See [CONTRIBUTING.md](CONTRIBUTING.md) and +[CHANGELOG.md](CHANGELOG.md) for project workflow and release notes. diff --git a/SECURITY.md b/SECURITY.md index c7ba04f..7987e2e 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -1,13 +1,38 @@ -# Security policy +# Security -Monium CLI v0.1.0 is intentionally offline. It does not authenticate to -Monium, store credentials, upload files, execute shell commands, or collect -telemetry. +This CLI is intentionally offline and credential-free. It does not collect or +transmit telemetry and does not accept or store tokens, cookies, or +credentials. It locally reads only the export files or stdin content that the +user explicitly supplies. Do not include secrets or sensitive export contents +in issues, logs, fixtures, or pull requests. -Do not include secrets or private log exports in bug reports. For a suspected -security issue, contact the repository maintainers privately before public -disclosure. Include a concise reproduction that uses synthetic identifiers -and omit credentials, tokens, cookies, and customer data. +The project does not implement or emulate the Monium web console's private +gateway. It does not scrape browser state, reuse console cookies, accept a +custom live endpoint, or fall back to opening a browser when a live operation +is unavailable. `monium capabilities` is local metadata only and performs no +network or credential probe. -The selector builders reject control characters and unsafe project/field names; -they are not a substitute for reviewing data before sharing a generated URL. +No dormant live provider, transport adapter, guessed alert schema, or public +provider/plugin ABI is packaged in this release. A future provider must be +based on a supported upstream API contract and official fixtures. Alert +ownership scope and telemetry query scope must be modeled separately; request +or response scope echoes are not treated as proof of server-side enforcement. + +Before registration, a provider must enforce request bytes/nodes before the +network boundary; stream response bodies under hard byte caps, absolute +connect/read/total deadlines, and cancellation; use a closed official alert +definition union; and expose honest completeness/redaction metadata. Unknown +schema semantics fail closed. Negative tests must cover cross-scope reads, +reserved filter overrides, list-to-read ID round trips, pagination faults, +oversized bodies, cancellation, and credential canaries in all output/error +paths. Custom endpoints, private gateways, browser sessions, and browser +fallbacks are prohibited for live operations. + +To report a security issue, use the repository's private advisory form: + + + +If private reporting is unavailable, open a non-sensitive GitHub issue without +reproduction details or secrets and ask the maintainers for a private contact +channel. Include only the minimum version and impact information needed to +start triage. diff --git a/pyproject.toml b/pyproject.toml index 8ce12c2..0999ded 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,19 +1,19 @@ [build-system] -requires = ["setuptools>=68"] +requires = ["setuptools==84.0.0"] build-backend = "setuptools.build_meta" [project] name = "monium-cli" -version = "0.1.0" -description = "An unofficial, offline-first CLI for Monium Logs and Traces workflows" +dynamic = ["version"] +description = "Unofficial, offline-first CLI for Yandex Monium logs and traces workflows" readme = "README.md" requires-python = ">=3.11" -license = { file = "LICENSE" } +license = "Apache-2.0" authors = [{ name = "Monium CLI contributors" }] +license-files = ["LICENSE"] keywords = ["monium", "observability", "logs", "traces", "cli"] classifiers = [ "Development Status :: 3 - Alpha", - "License :: OSI Approved :: Apache Software License", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", @@ -23,12 +23,24 @@ classifiers = [ ] dependencies = [] +[project.urls] +Homepage = "https://github.com/kizz-tech/monium" +Repository = "https://github.com/kizz-tech/monium" +Documentation = "https://github.com/kizz-tech/monium#readme" +Changelog = "https://github.com/kizz-tech/monium/blob/main/CHANGELOG.md" +Security = "https://github.com/kizz-tech/monium/security/policy" +"Yandex Monium documentation" = "https://yandex.cloud/docs/monium/" +Issues = "https://github.com/kizz-tech/monium/issues" + [project.scripts] monium = "monium_cli.cli:main" [tool.setuptools] package-dir = { "" = "src" } +[tool.setuptools.dynamic] +version = { attr = "monium_cli.__version__" } + [tool.setuptools.packages.find] where = ["src"] diff --git a/src/monium_cli/__init__.py b/src/monium_cli/__init__.py index 8cf8dd3..7998827 100644 --- a/src/monium_cli/__init__.py +++ b/src/monium_cli/__init__.py @@ -1,3 +1,4 @@ -"""Monium CLI package.""" +"""Unofficial, offline Yandex Monium CLI package.""" -__version__ = "0.1.0" +# Single version source: packaging reads this attribute via setuptools. +__version__ = "0.2.0rc1" diff --git a/src/monium_cli/cli.py b/src/monium_cli/cli.py index a90ba21..088ed15 100644 --- a/src/monium_cli/cli.py +++ b/src/monium_cli/cli.py @@ -1,4 +1,4 @@ -"""Command-line interface for the offline Monium CLI.""" +"""Command-line interface for the offline Monium toolkit.""" from __future__ import annotations @@ -7,8 +7,11 @@ import sys import webbrowser from collections.abc import Sequence +from typing import Any +from . import parser as _parser from . import __version__ +from .config import ConfigError, resolve_config from .parser import ExportError, parse_export from .selectors import ( DEFAULT_LOG_COLUMNS, @@ -20,15 +23,61 @@ ) +SCHEMA_VERSION = "1" + + +CAPABILITIES = ( + { + "name": "logs.selector", + "available": True, + "mode": "offline", + }, + { + "name": "logs.export.parse", + "available": True, + "mode": "offline", + }, + { + "name": "traces.selector", + "available": True, + "mode": "offline", + }, + { + "name": "alerts.list", + "available": False, + "mode": "live", + "reason_code": "no_supported_provider", + }, + { + "name": "alerts.read", + "available": False, + "mode": "live", + "reason_code": "no_supported_provider", + }, + { + "name": "logs.query.live", + "available": False, + "mode": "live", + "reason_code": "no_supported_provider", + }, +) + + class CliError(ValueError): - """A user-facing command-line error that should be rendered as JSON.""" + """A user-facing command-line error rendered as structured JSON.""" + def __init__(self, message: str, *, code: str = "cli_error") -> None: + super().__init__(message) + self.code = code -class JsonArgumentParser(argparse.ArgumentParser): - """Argparse parser whose validation errors are handled by :func:`main`.""" +class JsonArgumentParser(argparse.ArgumentParser): def error(self, message: str) -> None: # pragma: no cover - exercised via main - raise CliError(message) + # argparse messages may repeat unknown option values verbatim. Those + # values can contain credentials and must never reach structured + # stderr/stdout or incident logs. + del message + raise CliError("invalid command-line arguments; run monium --help", code="invalid_arguments") def _levels(values: list[list[str]] | None) -> list[str]: @@ -40,149 +89,245 @@ def _levels(values: list[list[str]] | None) -> list[str]: def _pairs(values: list[str] | None, option: str) -> list[tuple[str, str]]: - result = [] + return [parse_key_value(value, option) for value in (values or [])] + + +def _numeric(values: list[list[str]] | None, option: str = "--number") -> list[tuple[str, str, str]]: + result: list[tuple[str, str, str]] = [] + for group in values or []: + if len(group) != 3: + raise CliError(f"{option} expects FIELD OP VALUE", code="invalid_selector") + result.append((group[0], group[1], group[2])) + return result + + +def _numeric_equal(values: list[str] | None) -> list[tuple[str, str, str]]: + result: list[tuple[str, str, str]] = [] for value in values or []: - result.append(parse_key_value(value, option)) + if "=" not in value: + raise CliError("--number-field expects KEY=NUMBER", code="invalid_selector") + key, number = value.split("=", 1) + if not key.strip() or not number.strip(): + raise CliError("--number-field expects KEY=NUMBER", code="invalid_selector") + result.append((key.strip(), "=", number.strip())) return result +def _config_args(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--config", dest="config_path", default=argparse.SUPPRESS, help="optional non-secret TOML config") + parser.add_argument("--no-config", action="store_true", default=argparse.SUPPRESS, help="ignore the TOML config file") + + +def _selector_args(parser: argparse.ArgumentParser, *, project_required: bool = False) -> None: + parser.add_argument("--project", required=project_required, help="Monium project/scope identifier") + parser.add_argument("--service", help="service selector") + parser.add_argument("--log-group-id", help="log group identifier (alternative Logs scope)") + parser.add_argument("--cluster", help="cluster selector") + parser.add_argument("--host", help="host selector") + parser.add_argument("--level", help="exact log level") + parser.add_argument("--min-level", help="minimum log level") + parser.add_argument("--contains", dest="contains", help="substring on message") + parser.add_argument("--trace-id", help="trace identifier") + parser.add_argument("--span-id", help="span identifier") + parser.add_argument("--field", "--exact", action="append", dest="field", help="exact string field filter KEY=VALUE (repeatable)") + parser.add_argument("--contains-field", action="append", help="contains filter KEY=VALUE; field must be message or meta.*") + parser.add_argument("--glob-field", "--glob", action="append", dest="glob_field", help="glob string filter KEY=VALUE (repeatable)") + parser.add_argument("--regex-field", "--regex", action="append", dest="regex_field", help="regex string filter KEY=VALUE (repeatable)") + parser.add_argument("--number", action="append", nargs=3, metavar=("FIELD", "OP", "VALUE"), help="numeric comparison, e.g. --number latency_ms >= 500") + parser.add_argument("--number-field", action="append", help="numeric equality FIELD=NUMBER (repeatable)") + + def build_parser() -> argparse.ArgumentParser: parser = JsonArgumentParser( prog="monium", - description="Build Monium Logs/Traces links and parse exported NDJSON offline.", + description="Unofficial CLI for Yandex Monium (offline builders and parsers).", ) parser.add_argument("--version", action="version", version=__version__) + _config_args(parser) commands = parser.add_subparsers(dest="domain", required=True) - logs = commands.add_parser("logs", help="build Logs links or parse an export") + capabilities = commands.add_parser("capabilities", help="report supported operations without probing credentials or network") + capabilities.set_defaults(handler=_handle_capabilities) + + logs = commands.add_parser("logs", help="build Logs selectors/URLs or parse an export") + _config_args(logs) log_commands = logs.add_subparsers(dest="logs_command", required=True) - url = log_commands.add_parser("url", help="build a deterministic Monium Logs URL") - url.add_argument("--project", required=True, help="Monium project/scope identifier") - url.add_argument("--service", required=True, help="service selector") - url.add_argument("--cluster", help="cluster selector") - url.add_argument("--host", help="host selector") - url.add_argument("--level", help="exact log level (for example ERROR)") - url.add_argument("--min-level", help="minimum level (for example WARN)") - url.add_argument("--contains", "--text", "--message", dest="contains", help="substring to find in messages") - url.add_argument("--path", help="request path substring") - url.add_argument("--status-code", help="HTTP status code") - url.add_argument("--trace-id", help="trace identifier") - url.add_argument("--span-id", help="span identifier") - url.add_argument("--field", action="append", help="exact field filter, KEY=VALUE (repeatable)") - url.add_argument("--contains-field", action="append", help="substring field filter, KEY=VALUE (repeatable)") - url.add_argument("--from", "--from-time", dest="from_time", default="now-1h", help="Logs start time") - url.add_argument("--to", "--to-time", dest="to_time", default="now", help="Logs end time") - url.add_argument("--columns", nargs="+", default=list(DEFAULT_LOG_COLUMNS), help="columns to show") + url = log_commands.add_parser("url", help="build a deterministic Logs UI URL (volatile route)") + _config_args(url) + _selector_args(url) + url.add_argument("--from", "--from-time", dest="from_time", help="Logs start time") + url.add_argument("--to", "--to-time", dest="to_time", help="Logs end time") + url.add_argument("--columns", nargs="+", help="columns to show") url.add_argument("--tab", choices=("logs", "errors", "aggregates"), default="logs") - url.add_argument("--open", action="store_true", help="ask the system browser to open the generated URL") + url.add_argument("--ui-base-url", help="validated HTTPS UI origin") + url.add_argument("--open", action="store_true", help="opt-in: ask the system browser to open the URL") url.set_defaults(handler=_handle_logs_url) - parse = log_commands.add_parser("parse", help="parse a Monium/OpenTelemetry NDJSON export") - parse.add_argument("export", help="path to an NDJSON export") + query = log_commands.add_parser("query", help="build a Logs selector without contacting Monium") + _config_args(query) + _selector_args(query) + query.set_defaults(handler=_handle_logs_query) + + parse = log_commands.add_parser("parse", help="parse NDJSON, JSON, CSV, or text locally") + _config_args(parse) + parse.add_argument("export", help="path or - for stdin") + parse.add_argument("--input-format", choices=("auto", "ndjson", "json", "csv", "text"), default=None) parse.add_argument("--service", help="exact service filter") parse.add_argument("--cluster", help="exact cluster filter") - parse.add_argument( - "--level", - "--levels", - dest="levels", - action="append", - nargs="+", - help="level filter; repeat or separate values with commas", - ) + parse.add_argument("--level", "--levels", dest="levels", action="append", nargs="+", help="level filter; repeat or separate values with commas") parse.add_argument("--contains", "--text", "--message", dest="contains", help="message substring filter") parse.add_argument("--limit", type=int, default=20, help="maximum rows to return (default: 20)") parse.add_argument("--include-attributes", "--include-meta", dest="include_attributes", action="store_true") - parse.add_argument("--summary-only", action="store_true", help="omit rows while retaining the summary") + parse.add_argument("--summary-only", action="store_true", help="omit rows and mark rows_omitted while retaining summary") parse.set_defaults(handler=_handle_logs_parse) - traces = commands.add_parser("traces", help="build a deterministic Traces query") - trace_query = traces.add_subparsers(dest="traces_command", required=True) - query = trace_query.add_parser("query", help="build a Monium Traces selector query") - query.add_argument("--project", required=True, help="Monium project/scope identifier") - query.add_argument("--cluster", help="cluster selector") - query.add_argument("--service", help="service selector") - query.add_argument("--trace-id", help="trace identifier") - query.add_argument("--span-id", help="span identifier") - query.add_argument("--operation", "--span-name", dest="operation", help="operation/span name") - query.add_argument("--span-kind", help="span kind") - query.add_argument("--span-status", help="span status") - query.add_argument("--error-only", action="store_true", help="add span.status==ERROR") - query.add_argument("--min-duration", help="minimum duration, such as 1s") - query.add_argument("--max-duration", help="maximum duration, such as 5s") - query.add_argument("--attr", action="append", help="exact span attribute, KEY=VALUE (repeatable)") - query.add_argument("--attr-not", action="append", help="negative span attribute, KEY=VALUE (repeatable)") - query.add_argument("--attr-regex", action="append", help="regex span attribute, KEY=VALUE (repeatable)") - query.set_defaults(handler=_handle_traces_query) + traces = commands.add_parser("traces", help="build a deterministic Traces selector") + _config_args(traces) + trace_commands = traces.add_subparsers(dest="traces_command", required=True) + trace_query = trace_commands.add_parser("query", help="build a Traces selector query") + _config_args(trace_query) + trace_query.add_argument("--project", help="Monium project/scope identifier") + trace_query.add_argument("--cluster", help="cluster selector") + trace_query.add_argument("--service", help="service selector") + trace_query.add_argument("--trace-id", help="trace identifier") + trace_query.add_argument("--span-id", help="span identifier") + trace_query.add_argument("--operation", "--span-name", dest="operation", help="operation/span name") + trace_query.add_argument("--span-kind", help="span kind") + trace_query.add_argument("--span-status", help="span status") + trace_query.add_argument("--error-only", action="store_true", help="add span.status=ERROR") + trace_query.add_argument("--min-duration", help="minimum duration, such as 1s") + trace_query.add_argument("--max-duration", help="maximum duration, such as 5s") + trace_query.add_argument("--attr", action="append", help="exact span attribute KEY=VALUE (repeatable)") + trace_query.add_argument("--attr-not", action="append", help="negative span attribute KEY=VALUE (repeatable)") + trace_query.add_argument("--attr-regex", action="append", help="regex span attribute KEY=VALUE (repeatable)") + trace_query.add_argument("--critical-path", nargs="?", const="PRESENT", choices=("PRESENT", "ABSENT"), default=None, help="filter span.critical_path state (default PRESENT)") + trace_query.set_defaults(handler=_handle_traces_query) return parser -def _handle_logs_url(args: argparse.Namespace) -> dict[str, object]: - query = build_log_selector( - args.project, - args.service, - cluster=args.cluster, - host=args.host, - level=args.level, - min_level=args.min_level, - contains=args.contains, - path=args.path, - status_code=args.status_code, - trace_id=args.trace_id, - span_id=args.span_id, - fields=_pairs(args.field, "--field"), - contains_fields=_pairs(args.contains_field, "--contains-field"), - ) - url = build_logs_url( - args.project, - query, - from_time=args.from_time, - to_time=args.to_time, - columns=args.columns, - tab=args.tab, +def _handle_capabilities(args: argparse.Namespace) -> dict[str, object]: + del args + return { + "command": "capabilities", + "mode": "offline", + "capabilities": [dict(capability) for capability in CAPABILITIES], + "network_probed": False, + "credentials_probed": False, + } + + +def _resolve(args: argparse.Namespace) -> dict[str, Any]: + cli_values = { + key: getattr(args, key, None) + for key in ("project", "service", "cluster", "ui_base_url", "from_time", "to_time", "columns", "input_format") + } + cfg = resolve_config(config_path=getattr(args, "config_path", None), no_config=bool(getattr(args, "no_config", False)), cli_values=cli_values) + return cfg.values + + +def _log_selector(args: argparse.Namespace, config: dict[str, Any], *, require_project: bool = False) -> str: + project = getattr(args, "project", None) or config.get("project") + service = getattr(args, "service", None) or config.get("service") + cluster = getattr(args, "cluster", None) or config.get("cluster") + if require_project and project is None: + raise CliError("project is required (pass --project or configure project)", code="missing_project") + return build_log_selector( + project, + service, + cluster=cluster, + host=getattr(args, "host", None), + level=getattr(args, "level", None), + min_level=getattr(args, "min_level", None), + contains=getattr(args, "contains", None), + trace_id=getattr(args, "trace_id", None), + span_id=getattr(args, "span_id", None), + log_group_id=getattr(args, "log_group_id", None), + fields=_pairs(getattr(args, "field", None), "--field"), + contains_fields=_pairs(getattr(args, "contains_field", None), "--contains-field"), + glob_fields=_pairs(getattr(args, "glob_field", None), "--glob-field"), + regex_fields=_pairs(getattr(args, "regex_field", None), "--regex-field"), + numeric_fields=_numeric(getattr(args, "number", None)) + _numeric_equal(getattr(args, "number_field", None)), ) + + +def _effective_payload(config: dict[str, Any]) -> dict[str, Any]: + return {key: config[key] for key in sorted(config) if key in {"project", "service", "cluster", "ui_base_url", "from_time", "to_time", "columns", "input_format"}} + + +def _handle_logs_url(args: argparse.Namespace) -> dict[str, object]: + config = _resolve(args) + query = _log_selector(args, config, require_project=True) + project = getattr(args, "project", None) or config.get("project") + if project is None: + raise CliError("project is required", code="missing_project") + base = getattr(args, "ui_base_url", None) or config.get("ui_base_url") or "https://monium.yandex.cloud" + columns = getattr(args, "columns", None) or config.get("columns") or list(DEFAULT_LOG_COLUMNS) + from_time = getattr(args, "from_time", None) or config.get("from_time") or "now-1h" + to_time = getattr(args, "to_time", None) or config.get("to_time") or "now" + url = build_logs_url(project, query, from_time=from_time, to_time=to_time, columns=columns, tab=args.tab, base_url=base) + effective_config = {**config, "ui_base_url": base, "columns": columns, "from_time": from_time, "to_time": to_time} opened = False open_error = None if args.open: try: opened = bool(webbrowser.open(url)) - except Exception as exc: # pragma: no cover - platform/browser dependent - open_error = str(exc) + except Exception: # pragma: no cover - platform/browser dependent + open_error = "browser open failed" output: dict[str, object] = { "command": "logs url", - "project": args.project, + "mode": "offline", + "project": project, "query": query, "url": url, "open_requested": bool(args.open), "opened": opened, + "config": _effective_payload(effective_config), } if open_error: output["open_error"] = open_error return output +def _handle_logs_query(args: argparse.Namespace) -> dict[str, object]: + config = _resolve(args) + query = _log_selector(args, config) + return { + "command": "logs query", + "mode": "offline", + "project": getattr(args, "project", None) or config.get("project"), + "query": query, + "config": _effective_payload(config), + } + + def _handle_logs_parse(args: argparse.Namespace) -> dict[str, object]: + config = _resolve(args) result = parse_export( args.export, - service=args.service, - cluster=args.cluster, + service=args.service or config.get("service"), + cluster=args.cluster or config.get("cluster"), levels=_levels(args.levels), contains=args.contains, limit=args.limit, include_attributes=args.include_attributes, + input_format=args.input_format or config.get("input_format", "auto"), + summary_only=args.summary_only, ) - if args.summary_only: - result["rows"] = [] - result["returned"] = 0 - result["command"] = "logs parse" + effective_config = {**config, "input_format": result.get("input_format", args.input_format or config.get("input_format", "auto"))} + result.update({"command": "logs parse", "mode": "offline", "config": _effective_payload(effective_config)}) return result def _handle_traces_query(args: argparse.Namespace) -> dict[str, object]: + config = _resolve(args) + project = args.project or config.get("project") + if project is None: + raise CliError("project is required (pass --project or configure project)", code="missing_project") query = build_trace_query( - args.project, - cluster=args.cluster, - service=args.service, + project, + cluster=args.cluster or config.get("cluster"), + service=args.service or config.get("service"), trace_id=args.trace_id, span_id=args.span_id, operation=args.operation, @@ -194,12 +339,80 @@ def _handle_traces_query(args: argparse.Namespace) -> dict[str, object]: attributes=_pairs(args.attr, "--attr"), attributes_not=_pairs(args.attr_not, "--attr-not"), attributes_regex=_pairs(args.attr_regex, "--attr-regex"), + critical_path=args.critical_path, ) - return {"command": "traces query", "project": args.project, "query": query} + return {"command": "traces query", "mode": "offline", "project": project, "query": query, "config": _effective_payload(config)} + + +def _emit(payload: object) -> bool: + """Emit bounded machine-readable JSON. + + ``parse_export`` bounds its compact payload, while this function adds the + command envelope and pretty-print indentation. Measure the exact bytes + and node count that will be written so the public CLI cannot turn a + bounded parser result into an oversized stdout response. Oversized + success payloads fail closed with a small structured error and a non-zero + status from ``main``. + """ + + def encode(value: object, *, indent: int | None = 2) -> tuple[str, int, int]: + rendered = json.dumps(value, ensure_ascii=True, allow_nan=False, indent=indent, sort_keys=True) + # ``print`` below appends one newline; include it in the hard byte + # budget so measured bytes equal bytes written to stdout. + return rendered, len(rendered.encode("ascii")) + 1, _parser._json_node_count(value) + try: + rendered, output_bytes, output_nodes = encode(payload) + except (TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + rendered = "" + output_bytes = _parser.MAX_OUTPUT_BYTES + 1 + output_nodes = _parser.MAX_OUTPUT_NODES + 1 + + if output_bytes > _parser.MAX_OUTPUT_BYTES or output_nodes > _parser.MAX_OUTPUT_NODES: + code = "output_too_large" if output_bytes > _parser.MAX_OUTPUT_BYTES else "output_too_complex" + fallback: dict[str, object] = { + "ok": False, + "schema_version": SCHEMA_VERSION, + "tool_version": __version__, + "warnings": [], + "error_code": code, + "error": ( + f"output exceeds {_parser.MAX_OUTPUT_BYTES} bytes" + if code == "output_too_large" + else f"output exceeds {_parser.MAX_OUTPUT_NODES} JSON nodes" + ), + } + try: + rendered, fallback_bytes, fallback_nodes = encode(fallback) + except (TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + rendered, fallback_bytes, fallback_nodes = "", 0, 0 + # A test or embedding may deliberately set an impossibly small cap. + # Compact the error once before giving up; never write bytes over the + # configured limit. + if fallback_bytes > _parser.MAX_OUTPUT_BYTES or fallback_nodes > _parser.MAX_OUTPUT_NODES: + try: + rendered, fallback_bytes, fallback_nodes = encode(fallback, indent=None) + except (TypeError, ValueError, OverflowError, UnicodeError, RecursionError): + rendered, fallback_bytes, fallback_nodes = "", 0, 0 + if fallback_bytes > _parser.MAX_OUTPUT_BYTES or fallback_nodes > _parser.MAX_OUTPUT_NODES: + return False + try: + print(rendered) + except BrokenPipeError: # pragma: no cover - shell pipeline dependent + try: + sys.stdout.close() + finally: + return True + return False -def _emit(payload: object) -> None: - print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + try: + print(rendered) + except BrokenPipeError: # pragma: no cover - shell pipeline dependent + try: + sys.stdout.close() + finally: + return True + return True def main(argv: Sequence[str] | None = None) -> int: @@ -211,14 +424,15 @@ def main(argv: Sequence[str] | None = None) -> int: if handler is None: raise CliError("a command is required") payload = handler(args) - _emit({"ok": True, **payload}) - return 0 - except (CliError, SelectorError, ExportError) as exc: - _emit({"ok": False, "error": str(exc)}) + emitted = _emit({"ok": True, "schema_version": SCHEMA_VERSION, "tool_version": __version__, "warnings": [], **payload}) + return 0 if emitted else 2 + except (CliError, SelectorError, ConfigError, ExportError, ValueError, RecursionError) as exc: + code = getattr(exc, "code", "cli_error") + _emit({"ok": False, "schema_version": SCHEMA_VERSION, "tool_version": __version__, "warnings": [], "error_code": code, "error": str(exc)}) return 2 except KeyboardInterrupt: - _emit({"ok": False, "error": "interrupted"}) + _emit({"ok": False, "schema_version": SCHEMA_VERSION, "tool_version": __version__, "warnings": [], "error_code": "interrupted", "error": "interrupted"}) return 130 -__all__ = ["build_parser", "main"] +__all__ = ["CAPABILITIES", "build_parser", "main"] diff --git a/src/monium_cli/config.py b/src/monium_cli/config.py new file mode 100644 index 0000000..6d94a5d --- /dev/null +++ b/src/monium_cli/config.py @@ -0,0 +1,123 @@ +"""Optional, non-secret configuration for the offline CLI.""" + +from __future__ import annotations + +import os +import tomllib +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Mapping + + +class ConfigError(ValueError): + """Raised when configuration cannot be loaded or contains unsafe keys.""" + + def __init__(self, message: str, *, code: str = "config_error") -> None: + super().__init__(message) + self.code = code + + +ALLOWED_KEYS = frozenset({"project", "service", "cluster", "ui_base_url", "from_time", "to_time", "columns", "input_format"}) +ENV_KEYS = { + "MONIUM_PROJECT": "project", + "MONIUM_SERVICE": "service", + "MONIUM_CLUSTER": "cluster", + "MONIUM_UI_BASE_URL": "ui_base_url", + "MONIUM_FROM_TIME": "from_time", + "MONIUM_TO_TIME": "to_time", + "MONIUM_COLUMNS": "columns", + "MONIUM_INPUT_FORMAT": "input_format", +} + + +@dataclass(frozen=True) +class EffectiveConfig: + values: dict[str, Any] + source: str | None = None + + +def _validate_values(values: Mapping[str, Any], source: str) -> dict[str, Any]: + # ``source`` is retained in this private helper's signature so callers do + # not need to coordinate a separate validation API. It must never be + # interpolated into an error: for file-backed values it is user-controlled + # and may contain credentials or other sensitive path components. + del source + unknown = sorted(set(values) - ALLOWED_KEYS) + if unknown: + # Do not enumerate unknown keys. A TOML key is user input and can be + # crafted to contain a path, token, or another value that should not + # appear in structured CLI errors. + raise ConfigError("config contains unsupported key(s)", code="config_unknown_key") + result: dict[str, Any] = {} + for key, value in values.items(): + if key == "columns": + if isinstance(value, str): + value = [item.strip() for item in value.split(",") if item.strip()] + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + raise ConfigError("config columns must be a list of strings", code="config_invalid_value") + result[key] = list(value) + elif not isinstance(value, str): + raise ConfigError(f"config {key} must be a string", code="config_invalid_value") + else: + if len(value) > 1024 or any(ord(ch) < 0x20 for ch in value): + raise ConfigError(f"config {key} contains invalid characters", code="config_invalid_value") + result[key] = value + return result + + +def _load_file(path: str | Path) -> dict[str, Any]: + source = Path(path) + try: + with source.open("rb") as handle: + data = tomllib.load(handle) + except FileNotFoundError: + # Error messages intentionally omit the requested path. Besides + # avoiding noisy output, this prevents a credential-like argv value + # embedded in ``--config`` from reaching JSON, logs, or dashboards. + raise ConfigError("config file not found", code="config_missing") from None + except UnicodeDecodeError: + raise ConfigError("config file is not valid UTF-8", code="config_invalid_utf8") from None + except tomllib.TOMLDecodeError: + raise ConfigError("config file contains invalid TOML", code="config_invalid_toml") from None + except (OSError, ValueError): + # ``IsADirectoryError`` and permission failures are OSErrors; a path + # containing a NUL can surface as ValueError on some platforms. Keep + # one bounded, stable public code for all read failures and suppress + # the provider/OS exception text, which may echo the path. + raise ConfigError("config file could not be read", code="config_unreadable") from None + if not isinstance(data, dict): + raise ConfigError("config file must contain a table", code="config_invalid_value") + return _validate_values(data, "config file") + + +def resolve_config( + *, + config_path: str | Path | None = None, + no_config: bool = False, + cli_values: Mapping[str, Any] | None = None, + environ: Mapping[str, str] | None = None, +) -> EffectiveConfig: + """Resolve CLI > environment > TOML > defaults. + + No implicit home-directory config is loaded. A config file is read only + when explicitly named with ``--config``; this keeps invocation behaviour + deterministic and prevents accidental credential/config discovery. + """ + + values: dict[str, Any] = {} + source: str | None = None + if config_path is not None and not no_config: + values.update(_load_file(config_path)) + source = str(config_path) + env = environ if environ is not None else os.environ + for env_key, key in ENV_KEYS.items(): + raw = env.get(env_key) + if raw is not None and raw != "": + values[key] = [part.strip() for part in raw.split(",") if part.strip()] if key == "columns" else raw + values.update(_validate_values({k: v for k, v in (cli_values or {}).items() if v is not None}, "CLI")) + # Echo only allowlisted effective values. Unknown/credential-like key + # names are rejected by validation; values themselves are not secret-scanned. + return EffectiveConfig(values=values, source=source) + + +__all__ = ["ALLOWED_KEYS", "ConfigError", "EffectiveConfig", "resolve_config"] diff --git a/src/monium_cli/parser.py b/src/monium_cli/parser.py index f34918e..f5bc7c9 100644 --- a/src/monium_cli/parser.py +++ b/src/monium_cli/parser.py @@ -1,46 +1,66 @@ -"""Offline NDJSON parsing for common Monium and OpenTelemetry log exports.""" +"""Offline readers and normalisation for neutral log/telemetry exports. + +The parser has no Monium client and never opens a network connection. It +supports explicit NDJSON, JSON, CSV, and text readers plus a conservative +``auto`` mode. Only documented OpenTelemetry envelope keys are expanded; +ordinary record fields named ``data``, ``payload``, ``logs``, ``records``, +``items``, or ``value`` remain fields unless they are an unambiguous neutral +records envelope. +""" from __future__ import annotations +import csv +import datetime as _dt import json +import math import re +import sys +import threading from collections import Counter from collections.abc import Iterable, Iterator, Mapping from pathlib import Path -from typing import Any +from typing import Any, BinaryIO class ExportError(ValueError): - """Raised when an export cannot be read or contains malformed JSON.""" - + """Raised when an export cannot be read or contains malformed data.""" + + def __init__(self, message: str, *, code: str = "export_error", line: int | None = None, byte: int | None = None) -> None: + super().__init__(message) + self.code = code + self.line = line + self.byte = byte + + +MAX_LINE_BYTES = 4 * 1024 * 1024 +MAX_INPUT_BYTES = 64 * 1024 * 1024 +MAX_JSON_BYTES = 32 * 1024 * 1024 +MAX_DEPTH = 100 +MAX_INTEGER_DIGITS = 100 +MAX_SUMMARY_CARDINALITY = 10_000 +MAX_FIELDS = 2_000 +# ``parse_export`` returns a Python object, but callers normally serialise it +# as JSON immediately. Keep both budgets deterministic and enforce them +# before returning so a small input cannot amplify into an unbounded result +# (for example, a shared resource attribute copied onto every returned row). +MAX_OUTPUT_BYTES = 8 * 1024 * 1024 +MAX_OUTPUT_NODES = 1_000_000 +_CSV_FIELD_LIMIT_LOCK = threading.Lock() _MISSING = object() -_CONTAINER_KEYS = { - "records", - "logs", - "items", - "entries", - "resourcelogs", - "scopelogs", - "logrecords", - "results", - "data", -} -_WRAPPER_KEYS = {"record", "logrecord", "log", "entry", "payload"} _CANONICAL_MARKERS = { - "time", - "timestamp", - "observedtimestamp", - "observedtime", - "severity", - "severitytext", - "severitynumber", - "level", - "body", - "message", - "msg", - "traceid", - "spanid", + "time", "timestamp", "observedtimestamp", "observedtime", "timeunixnano", + "observedtimeunixnano", "severity", "severitytext", "severitynumber", "level", + "body", "message", "msg", "text", "traceid", "trace_id", "spanid", "span_id", +} +_KNOWN_RECORD_KEYS = { + "time", "timestamp", "observedtimestamp", "observedtime", "starttime", "timeunixnano", + "observedtimeunixnano", "level", "severity", "severitytext", "severitynumber", "body", + "message", "msg", "text", "service", "servicename", "cluster", "clustername", + "traceid", "trace_id", "traceid", "spanid", "span_id", "operation", "operationname", + "spankind", "spanstatus", "statuscode", "status_code", "requestpath", "request_path", + "attributes", "meta", "labels", "resourceattributes", "scopeattributes", "resource", "scope", } _SEVERITY_NUMBERS = { **{number: "TRACE" for number in range(1, 5)}, @@ -50,45 +70,150 @@ class ExportError(ValueError): **{number: "ERROR" for number in range(17, 21)}, **{number: "FATAL" for number in range(21, 25)}, } +_OTLP_CONTAINERS = {"resourcelogs", "scopelogs", "logrecords"} +# ``records`` is the sole neutral envelope name. Other generic names are +# always fields unless they are one of the documented OTLP containers above. +_NEUTRAL_CONTAINER_KEYS = {"records"} def _key_name(value: object) -> str: return re.sub(r"[^a-z0-9]", "", str(value).casefold()) -def _unwrap(value: Any) -> Any: - """Unwrap OpenTelemetry ``AnyValue`` objects without losing primitives.""" +def _reject_constant(value: str) -> Any: + raise ValueError(f"non-finite JSON number {value} is not allowed") + + +def _parse_int(value: str) -> int: + digits = value.lstrip("+-") + if len(digits) > MAX_INTEGER_DIGITS: + raise ValueError("integer is too large") + return int(value) + +def _parse_float(value: str) -> float: + parsed = float(value) + if not math.isfinite(parsed): + raise ValueError("non-finite JSON number is not allowed") + return parsed + + +def _decode_json(text: str, *, line: int | None = None, byte: int | None = None) -> Any: + try: + value = json.loads(text, parse_constant=_reject_constant, parse_int=_parse_int, parse_float=_parse_float) + _check_depth(value) + return value + except json.JSONDecodeError as exc: + # JSONDecodeError lineno/colno are character positions. Include the + # byte offset when available; line input is normally one NDJSON line. + err_line = (line or 1) + exc.lineno - 1 + err_byte = (byte or 0) + len(text[: exc.pos].encode("utf-8", "replace")) + raise ExportError(f"invalid JSON on line {err_line} byte {err_byte}: {exc.msg}", code="invalid_json", line=err_line, byte=err_byte) from exc + except (ValueError, RecursionError, OverflowError) as exc: + raise ExportError(f"invalid JSON near line {line or 1}: {exc}", code="invalid_json", line=line, byte=byte) from exc + + +def _check_depth(value: Any, depth: int = 0) -> None: + if depth > MAX_DEPTH: + raise ValueError(f"nested value exceeds maximum depth {MAX_DEPTH}") if isinstance(value, Mapping): - for key in ( - "stringValue", - "string_value", - "intValue", - "int_value", - "doubleValue", - "double_value", - "boolValue", - "bool_value", - "bytesValue", - "bytes_value", - "value", - ): - if key in value: - return _unwrap(value[key]) - # Some exporters use a lower-case key or a one-item AnyValue object. - for key, nested in value.items(): - if _key_name(key) in {"stringvalue", "intvalue", "doublevalue", "boolvalue", "bytesvalue", "value"}: - return _unwrap(nested) + for key, child in value.items(): + _check_depth(key, depth + 1) + _check_depth(child, depth + 1) + elif isinstance(value, list): + for child in value: + _check_depth(child, depth + 1) + + +def _anyvalue(value: Any) -> Any: + """Unwrap OpenTelemetry AnyValue scalar/array/kvlist forms.""" + + if isinstance(value, Mapping): + normalized = {_key_name(k): k for k in value} + scalar_keys = ( + "stringvalue", "intvalue", "doublevalue", "boolvalue", "bytesvalue", + "arrayvalue", "kvlistvalue", + ) + for key in scalar_keys: + original = normalized.get(key) + if original is not None: + nested = value[original] + if key == "arrayvalue": + if isinstance(nested, Mapping): + nested = next((v for k, v in nested.items() if _key_name(k) == "values"), nested) + return [_anyvalue(item) for item in (nested if isinstance(nested, list) else [nested])] + if key == "kvlistvalue": + if isinstance(nested, Mapping): + nested = next((v for k, v in nested.items() if _key_name(k) == "values"), nested) + result: dict[str, Any] = {} + if isinstance(nested, list): + for item in nested: + if isinstance(item, Mapping): + key_value = _mapping_value(item, ("key", "name")) + val_value = _mapping_value(item, ("value", "val")) + if key_value is not _MISSING: + result[str(_anyvalue(key_value))] = _anyvalue(val_value if val_value is not _MISSING else item) + elif isinstance(nested, Mapping): + result = {str(k): _anyvalue(v) for k, v in nested.items()} + return result + return _anyvalue(nested) + # ``{"value": ...}`` is a common neutral wrapper, but only unwrap it + # when it is the sole key so ordinary record fields stay intact. + if len(value) == 1 and "value" in value: + return _anyvalue(value["value"]) + return {str(k): _anyvalue(v) for k, v in value.items()} + if isinstance(value, list): + return [_anyvalue(item) for item in value] return value def _scalar(value: Any) -> Any: - value = _unwrap(value) - if isinstance(value, (str, int, float, bool)) or value is None: + value = _anyvalue(value) + if value is None or isinstance(value, (str, int, float, bool)): return value - if isinstance(value, (list, tuple, dict)): - return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) - return str(value) + return value + + +def _preserve(value: Any) -> Any: + """Copy an unknown field without interpreting it as an AnyValue envelope.""" + + if isinstance(value, Mapping): + return {str(key): _preserve(item) for key, item in value.items()} + if isinstance(value, list): + return [_preserve(item) for item in value] + return value + + +def _json_node_count(value: Any) -> int: + """Return a deterministic JSON node count for an output value. + + Object keys count as nodes as well as their values. Counting keys is + intentionally conservative: it bounds the concrete JSON representation, + not only the Python container count, and therefore remains safe for + large unknown-field maps. + """ + + if isinstance(value, Mapping): + return 1 + sum(1 + _json_node_count(child) for child in value.values()) + if isinstance(value, list): + return 1 + sum(_json_node_count(child) for child in value) + return 1 + + +def _json_size(value: Any) -> tuple[int, int]: + """Return canonical UTF-8 JSON bytes and node count for ``value``.""" + + try: + encoded = json.dumps( + value, + ensure_ascii=True, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ).encode("ascii") + return len(encoded), _json_node_count(value) + except (TypeError, ValueError, UnicodeError, OverflowError, RecursionError) as exc: + raise ExportError("output cannot be represented as bounded JSON", code="output_invalid") from exc def _mapping_value(mapping: Mapping[str, Any], aliases: Iterable[str]) -> Any: @@ -100,17 +225,11 @@ def _mapping_value(mapping: Mapping[str, Any], aliases: Iterable[str]) -> Any: def _attribute_items(value: Any) -> Iterator[tuple[str, Any]]: - """Yield ``(key, value)`` pairs from dict/list OTEL attribute forms.""" - - value = _unwrap(value) + value = _anyvalue(value) if isinstance(value, Mapping): - # An AnyValue object is a scalar, not an attribute map. - if any(_key_name(key) in {"stringvalue", "intvalue", "doublevalue", "boolvalue", "bytesvalue"} for key in value): - return for key, item in value.items(): yield str(key), _scalar(item) - return - if isinstance(value, list): + elif isinstance(value, list): for item in value: if not isinstance(item, Mapping): continue @@ -118,9 +237,7 @@ def _attribute_items(value: Any) -> Iterator[tuple[str, Any]]: if key is _MISSING: continue nested = _mapping_value(item, ("value", "val")) - if nested is _MISSING: - nested = item - yield str(_unwrap(key)), _scalar(nested) + yield str(_anyvalue(key)), _scalar(nested if nested is not _MISSING else item) def _collect_attributes(record: Mapping[str, Any]) -> dict[str, Any]: @@ -128,13 +245,11 @@ def _collect_attributes(record: Mapping[str, Any]) -> dict[str, Any]: for key, value in record.items(): normalized = _key_name(key) if normalized in {"attributes", "meta", "labels", "resourceattributes", "scopeattributes"}: - for attr_key, attr_value in _attribute_items(value): - attrs[attr_key] = attr_value + attrs.update(_attribute_items(value)) elif normalized in {"resource", "scope"} and isinstance(value, Mapping): - nested = _mapping_value(value, ("attributes", "resource.attributes", "resourceAttributes")) + nested = _mapping_value(value, ("attributes", "resource.attributes", "resourceAttributes", "scope.attributes", "scopeAttributes")) if nested is not _MISSING: - for attr_key, attr_value in _attribute_items(nested): - attrs[attr_key] = attr_value + attrs.update(_attribute_items(nested)) return attrs @@ -153,56 +268,105 @@ def _has_marker(record: Mapping[str, Any]) -> bool: return any(_key_name(key) in _CANONICAL_MARKERS for key in record) -def _iter_candidates(value: Any, inherited: Mapping[str, Any] | None = None) -> Iterator[tuple[Mapping[str, Any], Mapping[str, Any]]]: - """Flatten exporter envelopes while carrying resource attributes down.""" +def _looks_record(value: Any) -> bool: + return isinstance(value, Mapping) and _has_marker(value) + +def _iter_candidates(value: Any, inherited: Mapping[str, Any] | None = None, *, depth: int = 0) -> Iterator[tuple[Mapping[str, Any], Mapping[str, Any]]]: + """Flatten only explicit OTLP or unambiguous neutral envelopes.""" + + if depth > MAX_DEPTH: + raise ExportError(f"nested value exceeds maximum depth {MAX_DEPTH}", code="input_too_deep") if isinstance(value, list): for item in value: - yield from _iter_candidates(item, inherited) + yield from _iter_candidates(item, inherited, depth=depth + 1) return if not isinstance(value, Mapping): - raise ExportError("each NDJSON value must be an object or an array of objects") - + raise ExportError("each input value must be an object or an array of objects", code="record_shape") parent_attrs = dict(inherited or {}) parent_attrs.update(_collect_attributes(value)) expanded = False - for key, child in value.items(): normalized = _key_name(key) - if normalized in _CONTAINER_KEYS and isinstance(child, (Mapping, list)): - # A scalar ``data`` field is a valid log body, not an envelope. - if normalized == "data" and not isinstance(child, (Mapping, list)): - continue + if normalized in _OTLP_CONTAINERS and isinstance(child, (Mapping, list)): + expanded = True + yield from _iter_candidates(child, parent_attrs, depth=depth + 1) + elif normalized in _NEUTRAL_CONTAINER_KEYS and not _has_marker(value) and isinstance(child, list) and ( + (child and all(_looks_record(item) for item in child)) or (not child and len(value) == 1) + ): + # A neutral records envelope is accepted only when its children + # clearly look like records. Generic ``records``/``items`` fields + # are consequently preserved on ordinary records. expanded = True - yield from _iter_candidates(child, parent_attrs) - elif normalized in _WRAPPER_KEYS and isinstance(child, Mapping): + yield from _iter_candidates(child, parent_attrs, depth=depth + 1) + elif normalized in {"record", "logrecord", "entry", "log"} and not _has_marker(value) and _looks_record(child): expanded = True - yield from _iter_candidates(child, parent_attrs) - + yield from _iter_candidates(child, parent_attrs, depth=depth + 1) if not expanded or _has_marker(value): yield value, parent_attrs -def normalize_record(record: Mapping[str, Any], inherited_attributes: Mapping[str, Any] | None = None, *, line: int | None = None) -> dict[str, Any]: - """Convert one Monium/OpenTelemetry record to a compact stable row.""" +def _nanos_to_time(value: Any) -> Any: + value = _scalar(value) + if value is None: + return None + try: + if isinstance(value, bool): + return value + nanos = int(value) + if str(value).strip() != str(nanos): + return str(value) + dt = _dt.datetime.fromtimestamp(nanos / 1_000_000_000, tz=_dt.timezone.utc) + return dt.isoformat(timespec="microseconds").replace("+00:00", "Z") + except (TypeError, ValueError, OverflowError, OSError): + return str(value) + + +def _record_time(record: Mapping[str, Any], attrs: Mapping[str, Any]) -> Any: + for aliases in ( + ("time", "timestamp", "observedTimestamp", "observedTime", "startTime"), + ("timeUnixNano",), + ("observedTimeUnixNano",), + ): + value = _direct_or_attribute(record, attrs, aliases) + if value is not None: + return _nanos_to_time(value) if any("unixnano" in _key_name(alias) for alias in aliases) else value + return None + + +def normalize_record( + record: Mapping[str, Any], + inherited_attributes: Mapping[str, Any] | None = None, + *, + line: int | None = None, + byte: int | None = None, + input_format: str | None = None, +) -> dict[str, Any]: + """Convert one neutral/OTLP record to a stable JSON-compatible row.""" attrs = dict(inherited_attributes or {}) attrs.update(_collect_attributes(record)) - # OpenTelemetry records often place resource attrs in a sibling resource; - # the inherited map already contains them, while direct attrs win. level = _direct_or_attribute(record, attrs, ("level", "severityText", "severity")) + explicit_level = _mapping_value(record, ("level", "severityText")) + severity_value = _mapping_value(record, ("severity",)) + if explicit_level is _MISSING and severity_value is not _MISSING: + try: + numeric_severity = int(_scalar(severity_value)) + if str(_scalar(severity_value)).strip() == str(numeric_severity): + level = _SEVERITY_NUMBERS.get(numeric_severity, level) + except (TypeError, ValueError, OverflowError): + pass if level is None: - severity_number = _direct_or_attribute(record, attrs, ("severityNumber",)) + number = _direct_or_attribute(record, attrs, ("severityNumber",)) try: - level = _SEVERITY_NUMBERS.get(int(severity_number)) - except (TypeError, ValueError): + level = _SEVERITY_NUMBERS.get(int(number)) if number is not None else None + except (TypeError, ValueError, OverflowError): level = None if level is not None: level = str(level).upper() - - row = { + row: dict[str, Any] = { "line": line, - "time": _direct_or_attribute(record, attrs, ("time", "timestamp", "observedTimestamp", "observedTime", "startTime")), + "time": _record_time(record, attrs), "level": level, "service": _direct_or_attribute(record, attrs, ("service", "service.name", "serviceName")), "cluster": _direct_or_attribute(record, attrs, ("cluster", "cluster.name", "clusterName", "k8s.cluster.name")), @@ -213,17 +377,30 @@ def normalize_record(record: Mapping[str, Any], inherited_attributes: Mapping[st "status_code": _direct_or_attribute(record, attrs, ("status_code", "statusCode", "http.status_code", "httpStatusCode")), "request_path": _direct_or_attribute(record, attrs, ("request_path", "requestPath", "http.route", "url.path", "path")), } - if row["message"] is not None: - row["message"] = str(row["message"]) for field in ("time", "service", "cluster", "operation", "trace_id", "span_id", "request_path"): - if row[field] is not None: + if row[field] is not None and not isinstance(row[field], str): row[field] = str(row[field]) - # A status code is more useful as a number when the export carried one; - # retain an unusual value as text rather than dropping the row. + if row["message"] is not None and not isinstance(row["message"], str): + row["message"] = str(row["message"]) if isinstance(row["status_code"], float) and row["status_code"].is_integer(): row["status_code"] = int(row["status_code"]) - if inherited_attributes or _collect_attributes(record): + + unknown: dict[str, Any] = {} + for key, value in record.items(): + if _key_name(key) not in _KNOWN_RECORD_KEYS: + if len(unknown) >= MAX_FIELDS: + raise ExportError( + f"record contains more than {MAX_FIELDS} unknown fields", + code="too_many_fields", + line=line, + byte=byte, + ) + unknown[str(key)] = _preserve(value) + if unknown: + row["fields"] = dict(sorted(unknown.items(), key=lambda item: item[0])) + if attrs: row["attributes"] = dict(sorted(attrs.items(), key=lambda item: item[0])) + row["source"] = {"line": line, "byte": byte, "format": input_format} return row @@ -239,6 +416,136 @@ def _matches(row: Mapping[str, Any], *, service: str | None, cluster: str | None return True +def _format_for_path(path: str | Path, requested: str) -> str: + allowed = {"auto", "ndjson", "json", "csv", "text"} + if requested not in allowed: + raise ExportError(f"input format must be one of: {', '.join(sorted(allowed))}", code="invalid_input_format") + if requested != "auto": + return requested + if str(path) == "-": + raise ExportError("stdin requires --input-format when format is ambiguous", code="input_format_required") + suffix = Path(path).suffix.casefold() + if suffix in {".ndjson", ".jsonl"}: + return "ndjson" + if suffix == ".json": + return "json" + if suffix == ".csv": + return "csv" + if suffix in {".txt", ".log"}: + return "text" + # Unknown file extensions are handled conservatively using a small prefix; + # JSON-looking content is treated as JSON and line-delimited objects as + # NDJSON. Anything else is text. + try: + with Path(path).open("rb") as probe: + sample = probe.read(64 * 1024) + text = sample.decode("utf-8") + except (OSError, UnicodeError): + return "text" + nonempty = [line for line in text.splitlines() if line.strip()] + if nonempty and all(line.lstrip().startswith(("{", "[")) for line in nonempty[:2]): + if len(nonempty) > 1: + try: + for line in nonempty[:2]: + _decode_json(line) + return "ndjson" + except ExportError: + pass + if text.lstrip().startswith(("{", "[")): + return "json" + if "," in (nonempty[0] if nonempty else ""): + return "csv" + return "text" + + +def _open_binary(path: str | Path) -> tuple[BinaryIO, bool]: + if str(path) == "-": + stream = getattr(sys.stdin, "buffer", sys.stdin) + return stream, False + try: + return Path(path).open("rb"), True + except OSError as exc: + # ``str(exc)`` commonly includes the full path. Export paths are + # untrusted input and must not be echoed in structured CLI/API errors. + raise ExportError("cannot read export input", code="input_unreadable") from exc + + +def _as_bytes(value: bytes | str) -> bytes: + """Accept binary files and text-mode StringIO used by embedders/tests.""" + + return value.encode("utf-8") if isinstance(value, str) else value + + +class _BoundedCSVLines: + """Text iterator backed by bounded binary ``readline`` calls. + + ``csv.reader`` requests physical lines through ``__next__``. Keeping the + bound at this layer prevents an oversized line from being allocated before + validation, while still allowing quoted fields to span several bounded + physical lines. The current record start gives deterministic byte + provenance without relying on ``TextIOWrapper.tell``. Record provenance + is reset before each ``next(reader)`` by the caller and keeps only the + current record start (O(1) state). + """ + + def __init__(self, stream: Any, *, max_line_bytes: int | None = None, max_input_bytes: int | None = None) -> None: + self.stream = stream + self.max_line_bytes = MAX_LINE_BYTES if max_line_bytes is None else max_line_bytes + self.max_input_bytes = MAX_INPUT_BYTES if max_input_bytes is None else max_input_bytes + self.line_number = 0 + self.total_bytes = 0 + self.record_start_line: int | None = None + self.record_start_byte: int | None = None + + def begin_record(self) -> None: + """Prepare to capture the first physical line of the next CSV row.""" + + self.record_start_line = None + self.record_start_byte = None + + def __iter__(self) -> "_BoundedCSVLines": + return self + + def __next__(self) -> str: + try: + raw = self.stream.readline(self.max_line_bytes + 1) + except (OSError, ValueError) as exc: + # Never include a stream/path representation: custom file-like + # objects may put user-supplied secrets into their exception text. + raise ExportError("cannot read CSV input", code="input_unreadable") from exc + if raw in (b"", ""): + raise StopIteration + raw_bytes = _as_bytes(raw) + line_number = self.line_number + 1 + byte_offset = self.total_bytes + if len(raw_bytes) > self.max_line_bytes: + raise ExportError( + f"line {line_number} exceeds {self.max_line_bytes} bytes", + code="line_too_large", + line=line_number, + byte=byte_offset, + ) + self.total_bytes += len(raw_bytes) + if self.total_bytes > self.max_input_bytes: + raise ExportError(f"input exceeds {self.max_input_bytes} bytes", code="input_too_large", line=line_number, byte=byte_offset) + self.line_number = line_number + # The caller skips empty rows returned for blank physical lines. Do not + # attribute the following data record to those lines; quoted multiline + # fields retain the first non-empty physical line as their start. + if self.record_start_line is None and raw_bytes.rstrip(b"\r\n") != b"": + self.record_start_line = line_number + self.record_start_byte = byte_offset + try: + return raw_bytes.decode("utf-8") + except UnicodeDecodeError as exc: + raise ExportError( + f"export is not valid UTF-8 near line {line_number} byte {byte_offset + exc.start}", + code="invalid_utf8", + line=line_number, + byte=byte_offset + exc.start, + ) from exc + + def parse_export( path: str | Path, *, @@ -248,79 +555,295 @@ def parse_export( contains: str | None = None, limit: int = 20, include_attributes: bool = False, + input_format: str = "auto", + summary_only: bool = False, ) -> dict[str, Any]: - """Parse, filter, and summarize an NDJSON export. - - Invalid JSON is a hard error with a line number. This avoids silently - presenting a partial export as if it were complete. - """ + """Parse, filter, and summarize an export without network access.""" if limit < 0: - raise ExportError("limit must be zero or greater") + raise ExportError("limit must be zero or greater", code="invalid_limit") if limit > 100_000: - raise ExportError("limit must not exceed 100000") - level_values = {str(level).upper() for level in (levels or ()) if str(level).strip()} - if service is not None: - service = str(service) - if cluster is not None: - cluster = str(cluster) - - source = Path(path) - try: - handle = source.open("r", encoding="utf-8") - except OSError as exc: - raise ExportError(f"cannot read export {source}: {exc.strerror or exc}") from exc - - scanned = 0 - matched = 0 - line_number = 0 + raise ExportError("limit must not exceed 100000", code="invalid_limit") + fmt = _format_for_path(path, input_format) + level_values = {str(level).strip().upper() for level in (levels or ()) if str(level).strip()} + service = None if service is None else str(service) + cluster = None if cluster is None else str(cluster) + scanned = matched = 0 rows: list[dict[str, Any]] = [] levels_count: Counter[str] = Counter() services_count: Counter[str] = Counter() - try: - with handle: - for line_number, raw_line in enumerate(handle, start=1): - if not raw_line.strip(): + warnings: list[str] = [] + summary_cardinality_truncated = False + total_bytes = 0 + stream, should_close = _open_binary(path) + output_budget_exhausted = False + accepted_rows_bytes = 0 + accepted_rows_nodes = 0 + + def _payload_skeleton( + *, + returned: int, + rows_omitted: bool, + truncated: bool, + warning_values: list[str], + ) -> dict[str, Any]: + """Build an output shape with an empty rows list for budget admission.""" + + return { + "file": str(path), + "input_format": fmt, + "total_bytes": total_bytes, + "total_scanned": scanned, + "matched": matched, + "returned": returned, + "truncated": truncated, + "rows_omitted": rows_omitted, + "summary": { + "levels": dict(sorted(levels_count.items())), + "services": dict(sorted(services_count.items())), + }, + "warnings": list(warning_values), + "rows": [], + } + + def _mark_output_budget(reason: str) -> None: + nonlocal output_budget_exhausted + if output_budget_exhausted: + return + output_budget_exhausted = True + warnings.append(f"output {reason} budget reached; rows omitted") + + def consume(value: Any, *, line: int, byte: int) -> None: + nonlocal scanned, matched, summary_cardinality_truncated + nonlocal accepted_rows_bytes, accepted_rows_nodes + try: + candidates = _iter_candidates(value) + for candidate, inherited in candidates: + scanned += 1 + row = normalize_record(candidate, inherited, line=line, byte=byte, input_format=fmt) + if not _matches(row, service=service, cluster=cluster, levels=level_values, contains=contains): continue + matched += 1 + level_name = str(row.get("level") or "unknown") + service_name = str(row.get("service") or "unknown") + if level_name not in levels_count and len(levels_count) >= MAX_SUMMARY_CARDINALITY: + summary_cardinality_truncated = True + else: + levels_count[level_name] += 1 + if service_name not in services_count and len(services_count) >= MAX_SUMMARY_CARDINALITY: + summary_cardinality_truncated = True + else: + services_count[service_name] += 1 + if not summary_only and not output_budget_exhausted and len(rows) < limit: + if not include_attributes: + row.pop("attributes", None) + row_bytes, row_nodes = _json_size(row) + skeleton = _payload_skeleton( + returned=len(rows) + 1, + rows_omitted=False, + truncated=matched > limit, + warning_values=warnings, + ) + base_bytes, base_nodes = _json_size(skeleton) + # ``skeleton`` contains an empty rows list. Replace it + # with every accepted row plus this candidate, including + # one comma for each already accepted row. This keeps + # admission cumulative rather than accidentally checking + # each row in isolation. + candidate_bytes = base_bytes + accepted_rows_bytes + row_bytes + len(rows) + candidate_nodes = base_nodes + accepted_rows_nodes + row_nodes + if candidate_bytes > MAX_OUTPUT_BYTES: + _mark_output_budget("byte") + elif candidate_nodes > MAX_OUTPUT_NODES: + _mark_output_budget("JSON node") + else: + rows.append(row) + accepted_rows_bytes += row_bytes + accepted_rows_nodes += row_nodes + except ExportError: + raise + except (TypeError, ValueError, OverflowError, RecursionError) as exc: + raise ExportError(f"unsupported record shape on line {line}: {exc}", code="record_shape", line=line, byte=byte) from exc + + try: + if fmt == "json": + data = _as_bytes(stream.read(MAX_JSON_BYTES + 1)) + total_bytes += len(data) + if len(data) > MAX_JSON_BYTES: + raise ExportError(f"JSON input exceeds {MAX_JSON_BYTES} bytes", code="input_too_large") + try: + text = data.decode("utf-8") + except UnicodeDecodeError as exc: + raise ExportError(f"export is not valid UTF-8 near line 1 byte {exc.start}", code="invalid_utf8", line=1, byte=exc.start) from exc + consume(_decode_json(text, line=1, byte=0), line=1, byte=0) + elif fmt == "csv": + source = _BoundedCSVLines(stream) + # ``csv.field_size_limit`` is process-global in the stdlib. Keep + # the whole parse under a lock and always restore the caller's + # limit so concurrent parses cannot interleave settings. + with _CSV_FIELD_LIMIT_LOCK: + previous_field_limit = csv.field_size_limit() + csv.field_size_limit(source.max_line_bytes) try: - decoded = json.loads(raw_line) - except json.JSONDecodeError as exc: - raise ExportError(f"invalid JSON on line {line_number}: {exc.msg}") from exc + try: + reader = csv.reader(source, strict=True) + # Consume blank physical lines before the header + # explicitly; each ``next`` remains bounded by source. + fieldnames = None + while fieldnames is None: + source.begin_record() + try: + candidate = next(reader) + except StopIteration: + break + if candidate: + fieldnames = candidate + if fieldnames is None: + total_bytes = source.total_bytes + empty_payload = { + "file": str(path), + "input_format": fmt, + "total_bytes": total_bytes, + "total_scanned": 0, + "matched": 0, + "returned": 0, + "truncated": False, + "rows_omitted": bool(summary_only), + "summary": {"levels": {}, "services": {}}, + "warnings": [], + "rows": [], + } + output_bytes, output_nodes = _json_size(empty_payload) + if output_bytes > MAX_OUTPUT_BYTES: + raise ExportError(f"output exceeds {MAX_OUTPUT_BYTES} bytes", code="output_too_large") + if output_nodes > MAX_OUTPUT_NODES: + raise ExportError(f"output exceeds {MAX_OUTPUT_NODES} JSON nodes", code="output_too_complex") + return empty_payload + fieldnames = [str(field) for field in fieldnames] + if any(not field for field in fieldnames): + raise ExportError( + "CSV header names must not be empty", + code="invalid_csv", + line=source.record_start_line or 1, + byte=source.record_start_byte or 0, + ) + if len(set(fieldnames)) != len(fieldnames): + raise ExportError( + "CSV header names must be unique", + code="invalid_csv", + line=source.record_start_line or 1, + byte=source.record_start_byte or 0, + ) + while True: + source.begin_record() + try: + record = next(reader) + except StopIteration: + break + if not record: + continue + line = source.record_start_line or reader.line_num or 1 + byte = source.record_start_byte or 0 + if len(record) != len(fieldnames): + raise ExportError( + f"CSV row at line {line} has {len(record)} columns; expected {len(fieldnames)}", + code="invalid_csv", + line=line, + byte=byte, + ) + mapped = {str(field): record[index] for index, field in enumerate(fieldnames)} + consume(mapped, line=line, byte=byte) + except csv.Error as exc: + line = getattr(reader, "line_num", None) if "reader" in locals() else source.line_number + raise ExportError(f"invalid CSV near line {line or 1}: {exc}", code="invalid_csv", line=line or 1) from exc + finally: + csv.field_size_limit(previous_field_limit) + total_bytes = source.total_bytes + elif fmt == "text": + line_number = 0 + byte_offset = 0 + while True: + raw = stream.readline(MAX_LINE_BYTES + 1) + if raw in (b"", ""): + break + raw = _as_bytes(raw) + line_number += 1 + total_bytes += len(raw) + if total_bytes > MAX_INPUT_BYTES: + raise ExportError(f"input exceeds {MAX_INPUT_BYTES} bytes", code="input_too_large") + if len(raw) > MAX_LINE_BYTES: + raise ExportError(f"line {line_number} exceeds {MAX_LINE_BYTES} bytes", code="line_too_large", line=line_number, byte=byte_offset) try: - candidates = _iter_candidates(decoded) - for candidate, inherited in candidates: - scanned += 1 - row = normalize_record(candidate, inherited, line=line_number) - if not _matches(row, service=service, cluster=cluster, levels=level_values, contains=contains): - continue - matched += 1 - level = str(row.get("level") or "unknown") - service_name = str(row.get("service") or "unknown") - levels_count[level] += 1 - services_count[service_name] += 1 - if len(rows) < limit: - if not include_attributes: - row.pop("attributes", None) - rows.append(row) - except ExportError: - raise - except (TypeError, ValueError, OverflowError) as exc: - raise ExportError(f"unsupported record shape on line {line_number}: {exc}") from exc + text = raw.decode("utf-8") + except UnicodeDecodeError as exc: + raise ExportError(f"export is not valid UTF-8 near line {line_number} byte {byte_offset + exc.start}", code="invalid_utf8", line=line_number, byte=byte_offset + exc.start) from exc + message = text.rstrip("\r\n") + if message: + consume({"message": message}, line=line_number, byte=byte_offset) + byte_offset += len(raw) + else: # ndjson + line_number = 0 + byte_offset = 0 + while True: + raw = stream.readline(MAX_LINE_BYTES + 1) + if raw in (b"", ""): + break + raw = _as_bytes(raw) + line_number += 1 + total_bytes += len(raw) + if total_bytes > MAX_INPUT_BYTES: + raise ExportError(f"input exceeds {MAX_INPUT_BYTES} bytes", code="input_too_large") + if len(raw) > MAX_LINE_BYTES: + raise ExportError(f"line {line_number} exceeds {MAX_LINE_BYTES} bytes", code="line_too_large", line=line_number, byte=byte_offset) + try: + text = raw.decode("utf-8") + except UnicodeDecodeError as exc: + raise ExportError(f"export is not valid UTF-8 near line {line_number} byte {byte_offset + exc.start}", code="invalid_utf8", line=line_number, byte=byte_offset + exc.start) from exc + if text.strip(): + consume(_decode_json(text, line=line_number, byte=byte_offset), line=line_number, byte=byte_offset) + byte_offset += len(raw) + except OSError as exc: + raise ExportError("cannot read export input", code="input_unreadable") from exc except UnicodeError as exc: - raise ExportError(f"export is not valid UTF-8 near line {line_number or 1}") from exc - - return { - "file": str(source), + raise ExportError(f"export is not valid UTF-8", code="invalid_utf8") from exc + finally: + if should_close: + stream.close() + + if summary_cardinality_truncated: + warnings.append("summary cardinality exceeded the configured bound") + payload = { + "file": str(path), + "input_format": fmt, + "total_bytes": total_bytes, "total_scanned": scanned, "matched": matched, "returned": len(rows), - "truncated": matched > len(rows), + "truncated": matched > limit or output_budget_exhausted, + "rows_omitted": bool(summary_only or output_budget_exhausted), "summary": { "levels": dict(sorted(levels_count.items())), "services": dict(sorted(services_count.items())), }, + "warnings": warnings, "rows": rows, } - - -__all__ = ["ExportError", "normalize_record", "parse_export"] + output_bytes, output_nodes = _json_size(payload) + if output_bytes > MAX_OUTPUT_BYTES: + raise ExportError(f"output exceeds {MAX_OUTPUT_BYTES} bytes", code="output_too_large") + if output_nodes > MAX_OUTPUT_NODES: + raise ExportError(f"output exceeds {MAX_OUTPUT_NODES} JSON nodes", code="output_too_complex") + return payload + + +__all__ = [ + "ExportError", + "MAX_INPUT_BYTES", + "MAX_JSON_BYTES", + "MAX_LINE_BYTES", + "MAX_OUTPUT_BYTES", + "MAX_OUTPUT_NODES", + "MAX_SUMMARY_CARDINALITY", + "normalize_record", + "parse_export", +] diff --git a/src/monium_cli/selectors.py b/src/monium_cli/selectors.py index f81f1f1..026a89a 100644 --- a/src/monium_cli/selectors.py +++ b/src/monium_cli/selectors.py @@ -1,14 +1,15 @@ -"""Pure builders for Monium Logs and Traces selectors. +"""Offline, typed builders for Monium Logs and Traces selectors. -The builders intentionally do not contact Monium. They only validate values, -escape selector literals, and construct stable query strings and URLs that can -be pasted into a browser or consumed by automation. +This module deliberately contains no network or authentication code. The +public builders validate a small selector IR and render deterministic strings +which can be copied into the (volatile) Monium web UI. """ from __future__ import annotations import re import urllib.parse +from dataclasses import dataclass from collections.abc import Iterable, Sequence @@ -19,12 +20,63 @@ class SelectorError(ValueError): _PROJECT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,199}$") _KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_.:-]{0,127}$") _TOKEN_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.:-]{0,63}$") -_NUMBER_RE = re.compile(r"^-?(?:\d+\.?\d*|\d*\.\d+)$") -_DURATION_RE = re.compile(r"^(?:\d+\.?\d*|\d*\.\d+)(?:ns|us|µs|ms|s|m|h)$") +_NUMBER_RE = re.compile(r"^-?(?:\d+(?:\.\d*)?|\.\d+)$") +# The unified duration grammar accepts microseconds (``us``), milliseconds, +# seconds, minutes, and hours. Nanoseconds, the micro sign, and days are not +# accepted by the selector grammar used by this tool. +_DURATION_RE = re.compile(r"^(?:\d+(?:\.\d*)?|\.\d+)(?:us|ms|s|m|h)$") _CONTROL_RE = re.compile(r"[\x00-\x1f\x7f]") DEFAULT_LOG_COLUMNS = ("level", "time", "message", "host") LOG_TABS = frozenset(("logs", "errors", "aggregates")) +ALLOWED_OPERATORS = frozenset(("=", "==", "!=", "=*", "=~", ">", ">=", "<", "<=")) + +# These fields are rendered by the typed selector arguments above. Generic +# filters must not be able to add a second clause for them: duplicate clauses +# would make the meaning of a scope/target selector depend on provider-specific +# precedence rules instead of the CLI's explicit arguments. +_LOG_RESERVED_GENERIC_FIELDS = frozenset( + ("project", "cluster", "service", "log_group_id", "trace.id", "span.id") +) +_TRACE_RESERVED_ATTRIBUTE_FIELDS = frozenset( + ( + "project", + "cluster", + "service", + "trace.id", + "span.id", + "span.name", + "span.kind", + "span.status", + "span.duration", + "span.critical_path", + ) +) +_RESERVED_SELECTOR_ERROR = "generic filters cannot target first-class selector field" + + +@dataclass(frozen=True) +class SelectorClause: + """A typed, validated selector clause. + + ``value_type`` controls literal rendering. In particular, trace IDs and + statuses are always strings even when their spelling looks numeric or + boolean; only values explicitly marked ``number`` or ``duration`` are + rendered without quotes. + """ + + field: str + operator: str + value: str + value_type: str = "string" + + def __post_init__(self) -> None: + validate_key(self.field, "selector field") + if self.operator not in ALLOWED_OPERATORS: + raise SelectorError(f"unsupported selector operator: {self.operator}") + validate_value(self.value, "selector value") + if self.value_type not in {"string", "token", "number", "duration", "glob", "regex"}: + raise SelectorError(f"unsupported selector value type: {self.value_type}") def _text(value: object, name: str, *, max_length: int = 512, allow_empty: bool = False) -> str: @@ -43,8 +95,6 @@ def _text(value: object, name: str, *, max_length: int = 512, allow_empty: bool def validate_project(project: object) -> str: - """Validate a project/scope identifier used in a URL path and selector.""" - value = _text(project, "project", max_length=200) if not _PROJECT_RE.fullmatch(value): raise SelectorError( @@ -71,18 +121,18 @@ def validate_value(value: object, name: str = "value", *, max_length: int = 512) def validate_duration(value: object, name: str = "duration") -> str: text = validate_value(value, name, max_length=64) if not _DURATION_RE.fullmatch(text): - raise SelectorError(f"{name} must be a non-negative duration such as 500ms, 2s, or 1.5m") + raise SelectorError(f"{name} must be a non-negative duration such as 150us, 500ms, 2s, or 1.5m") return text -def quote_log_value(value: object) -> str: - """Return a Monium log-selector string literal. +def validate_number(value: object, name: str = "number") -> str: + text = validate_value(value, name, max_length=64) + if not _NUMBER_RE.fullmatch(text): + raise SelectorError(f"{name} must be a decimal number") + return text - Only a small, explicit escape set is used. The input is validated before - this function is called, so selector punctuation in a value cannot escape - its quoted literal. - """ +def quote_log_value(value: object) -> str: text = validate_value(value) escaped = ( text.replace("\\", "\\\\") @@ -94,42 +144,78 @@ def quote_log_value(value: object) -> str: return f'"{escaped}"' -def _render_selector_value(value: object, *, raw: bool = False) -> str: +def _render_literal(value: object, *, value_type: str = "string") -> str: text = validate_value(value) - if raw: + if value_type in {"number", "duration", "token"}: + if value_type == "number": + return validate_number(text) + if value_type == "duration": + return validate_duration(text) if not _TOKEN_RE.fullmatch(text): - raise SelectorError("raw selector values must be simple tokens") + raise SelectorError("token selector values must be simple identifiers") return text return quote_log_value(text) def parse_key_value(item: str, option: str = "--field") -> tuple[str, str]: - """Parse and validate an explicit ``KEY=VALUE`` filter.""" + """Parse an explicit ``KEY=VALUE`` pair without interpreting operators.""" - if "=" not in item: + if not isinstance(item, str) or "=" not in item: raise SelectorError(f"{option} expects KEY=VALUE") key, value = item.split("=", 1) return validate_key(key.strip(), f"{option} key"), validate_value(value.strip(), f"{option} value") -def _append_selector( - selectors: list[tuple[str, str, str, bool]], - key: str, - value: object, - *, - operator: str = "==", - raw: bool = False, -) -> None: - if value is None or value == "": - return - if operator not in {"=", "==", "!=", ">=", "<=", "=*", "=~"}: - raise SelectorError(f"unsupported selector operator: {operator}") - selectors.append((validate_key(key), operator, validate_value(value), raw)) +def parse_clause(item: str, option: str = "--clause") -> SelectorClause: + """Parse a documented ``FIELD OP VALUE`` clause. + + This parser is intentionally strict and only accepts one of the known + operators. Values are always strings; callers that need numeric or + duration semantics should use the typed flags/builders below. + """ + + if not isinstance(item, str): + raise SelectorError(f"{option} expects FIELD OP VALUE") + match = re.fullmatch(r"\s*([^\s]+)\s*(==|!=|>=|<=|=\*|=~|=|>|<)\s*(.+?)\s*", item) + if not match: + raise SelectorError(f"{option} expects FIELD OP VALUE using a supported operator") + key, operator, value = match.groups() + return SelectorClause(validate_key(key, f"{option} field"), operator, validate_value(value, f"{option} value")) + + +def _clause(field: object, value: object, operator: str, *, value_type: str = "string") -> SelectorClause: + return SelectorClause(validate_key(field), operator, validate_value(value), value_type) + + +def _render_clauses(clauses: Iterable[SelectorClause]) -> str: + rendered = [f"{c.field}{c.operator}{_render_literal(c.value, value_type=c.value_type)}" for c in clauses] + return "{ " + ", ".join(rendered) + " }" + + +def _validate_generic_field(field: object, *, family: str, reserved: frozenset[str]) -> str: + """Validate a generic field and reject first-class selector overrides. + + ``family`` is included in the validation context so malformed keys retain + useful diagnostics, while reserved-field failures intentionally share one + stable error contract across Logs and Traces generic filter families. + """ + + key = validate_key(field, f"{family} field") + if key in reserved: + raise SelectorError(f"{_RESERVED_SELECTOR_ERROR}: {key}") + return key + + +def _validate_contains_field(field: object) -> str: + key = _validate_generic_field(field, family="contains", reserved=_LOG_RESERVED_GENERIC_FIELDS) + if key != "message" and not (key.startswith("meta.") and len(key) > len("meta.")): + raise SelectorError("contains filters are only allowed on message or meta.* fields") + return key def build_log_selector( - project: object, - service: object, + project: object | None, + service: object | None = None, *, cluster: object | None = None, host: object | None = None, @@ -137,57 +223,141 @@ def build_log_selector( min_level: object | None = None, contains: object | None = None, message: object | None = None, - path: object | None = None, - status_code: object | None = None, trace_id: object | None = None, span_id: object | None = None, + log_group_id: object | None = None, fields: Iterable[tuple[str, str]] | None = None, contains_fields: Iterable[tuple[str, str]] | None = None, + glob_fields: Iterable[tuple[str, str]] | None = None, + regex_fields: Iterable[tuple[str, str]] | None = None, + numeric_fields: Iterable[tuple[str, str, object]] | None = None, + clauses: Iterable[SelectorClause] | None = None, ) -> str: - """Build a deterministic Logs selector. + """Build a deterministic Logs selector from typed generic clauses. - ``contains`` and ``message`` are aliases. ``fields`` and - ``contains_fields`` are already split ``KEY=VALUE`` pairs; use - :func:`parse_key_value` for command-line values. + A Logs query accepts project+service, a log-group ID alone, or a trace ID + alone. ``logs url`` separately requires project for its UI route. There + are intentionally no domain-specific request-path or status-code + shortcuts; use typed generic field filters when those fields exist in an + export. """ - project_value = validate_project(project) - service_value = validate_value(service, "service") + project_value = None if project is None else validate_project(project) + service_value = None if service is None else validate_value(service, "service") + group_value = None if log_group_id is None else validate_value(log_group_id, "log-group-id") + trace_value = None if trace_id is None else validate_value(trace_id, "trace-id") if contains is not None and message is not None: raise SelectorError("pass only one of contains and message") message_value = contains if contains is not None else message - - selectors: list[tuple[str, str, str, bool]] = [] - _append_selector(selectors, "project", project_value, operator="==") - _append_selector(selectors, "cluster", cluster, operator="==") - _append_selector(selectors, "service", service_value, operator="==") - _append_selector(selectors, "host", host, operator="==") + if level is not None and min_level is not None: + raise SelectorError("--level and --min-level cannot be used together") + if group_value is None and trace_value is None and (project_value is None or service_value is None): + raise SelectorError("logs query requires project plus service, log-group-id, or trace-id") + + result: list[SelectorClause] = [] + if project_value is not None: + result.append(_clause("project", project_value, "==")) + if cluster is not None: + result.append(_clause("cluster", cluster, "==")) + if service_value is not None: + result.append(_clause("service", service_value, "==")) + if group_value is not None: + result.append(_clause("log_group_id", group_value, "==")) + if host is not None: + result.append(_clause("host", host, "==")) if level is not None: - level_value = validate_value(level, "level", max_length=64).upper() - _append_selector(selectors, "level", level_value, operator="==", raw=True) + result.append(_clause("level", validate_value(level, "level", max_length=64).upper(), "==", value_type="token")) if min_level is not None: - min_level_value = validate_value(min_level, "min-level", max_length=64).upper() - _append_selector(selectors, "level", min_level_value, operator=">=", raw=True) + result.append(_clause("level", validate_value(min_level, "min-level", max_length=64).upper(), ">=", value_type="token")) if message_value is not None: - _append_selector(selectors, "message", message_value, operator="=*") - if path is not None: - _append_selector(selectors, "meta.RequestPath", path, operator="=*") - if status_code is not None: - status = str(status_code).strip() - if not re.fullmatch(r"\d{3}", status): - raise SelectorError("status-code must be a three-digit HTTP status") - _append_selector(selectors, "meta.StatusCode", status, raw=True) - _append_selector(selectors, "trace.id", trace_id, operator="==") - _append_selector(selectors, "span.id", span_id, operator="==") + result.append(_clause("message", message_value, "=*", value_type="glob")) + if trace_value is not None: + result.append(_clause("trace.id", trace_value, "==")) + if span_id is not None: + result.append(_clause("span.id", span_id, "==")) for key, value in fields or (): - _append_selector(selectors, key, value, operator="==") + result.append( + _clause( + _validate_generic_field(key, family="exact", reserved=_LOG_RESERVED_GENERIC_FIELDS), + value, + "==", + ) + ) for key, value in contains_fields or (): - _append_selector(selectors, key, value, operator="=*") + result.append(_clause(_validate_contains_field(key), value, "=*", value_type="glob")) + for key, value in glob_fields or (): + result.append( + _clause( + _validate_generic_field(key, family="glob", reserved=_LOG_RESERVED_GENERIC_FIELDS), + value, + "=", + value_type="glob", + ) + ) + for key, value in regex_fields or (): + result.append( + _clause( + _validate_generic_field(key, family="regex", reserved=_LOG_RESERVED_GENERIC_FIELDS), + value, + "=~", + value_type="regex", + ) + ) + for key, operator, value in numeric_fields or (): + validated_key = _validate_generic_field(key, family="numeric", reserved=_LOG_RESERVED_GENERIC_FIELDS) + if operator not in {"=", "!=", ">", ">=", "<", "<="}: + raise SelectorError("numeric filters support =, !=, >, >=, <, <=") + result.append( + _clause( + validated_key, + validate_number(value, "numeric value"), + operator, + value_type="number", + ) + ) + for clause in clauses or (): + _validate_generic_field(clause.field, family="clause", reserved=_LOG_RESERVED_GENERIC_FIELDS) + result.append(clause) + return _render_clauses(result) - rendered = [] - for key, operator, value, raw in selectors: - rendered.append(f"{key}{operator}{_render_selector_value(value, raw=raw)}") - return "{ " + ", ".join(rendered) + " }" + +def validate_https_origin(base_url: object) -> str: + """Validate an HTTPS origin for the volatile UI adapter. + + Userinfo, whitespace/control characters, paths, queries, fragments, and + malformed/out-of-range ports are rejected. A trailing slash is tolerated + and removed for deterministic output. + """ + + if base_url is None: + raise SelectorError("base URL is required") + raw_text = str(base_url) + if raw_text != raw_text.strip(): + raise SelectorError("base URL must not contain whitespace or control characters") + text = _text(raw_text, "base URL", max_length=512) + if any(ch.isspace() or ord(ch) < 0x20 or ord(ch) == 0x7F for ch in text): + raise SelectorError("base URL must not contain whitespace or control characters") + try: + parsed = urllib.parse.urlsplit(text) + except ValueError as exc: + raise SelectorError("base URL must be an HTTPS origin") from exc + if parsed.scheme.lower() != "https": + raise SelectorError("base URL must be an HTTPS origin") + if parsed.username is not None or parsed.password is not None: + raise SelectorError("base URL must not contain userinfo") + if not parsed.hostname: + raise SelectorError("base URL must include a hostname") + try: + port = parsed.port + except ValueError as exc: + raise SelectorError("base URL has an invalid port") from exc + if parsed.netloc.endswith(":") or port is not None and not (1 <= port <= 65535): + raise SelectorError("base URL has an invalid port") + if parsed.path not in {"", "/"} or parsed.query or parsed.fragment: + raise SelectorError("base URL must be an origin without path, query, or fragment") + # Preserve IPv6 brackets and explicit port while canonicalising a trailing + # slash. urlsplit's netloc is safe after the checks above. + return f"https://{parsed.netloc}" def build_logs_url( @@ -200,8 +370,6 @@ def build_logs_url( tab: str = "logs", base_url: str = "https://monium.yandex.cloud", ) -> str: - """Build the stable Log Explorer URL used by ``logs url``.""" - project_value = validate_project(project) query_value = validate_value(query, "query", max_length=20_000) from_value = validate_value(from_time, "from-time", max_length=128) @@ -210,13 +378,8 @@ def build_logs_url( raise SelectorError(f"tab must be one of: {', '.join(sorted(LOG_TABS))}") if not columns: raise SelectorError("at least one column is required") - normalized_columns = [] - for column in columns: - normalized_columns.append(validate_key(column, "column")) - - base = str(base_url).rstrip("/") - if not re.fullmatch(r"https://[^/?#]+", base): - raise SelectorError("base URL must be an HTTPS origin") + normalized_columns = [validate_key(column, "column") for column in columns] + base = validate_https_origin(base_url) params = { "query": query_value, "from": from_value, @@ -228,11 +391,8 @@ def build_logs_url( return f"{base}/projects/{path_project}/logs?{urllib.parse.urlencode(params)}" -def _trace_literal(value: object) -> str: - text = validate_value(value) - if text in {"true", "false"} or _NUMBER_RE.fullmatch(text) or _DURATION_RE.fullmatch(text): - return text - return quote_log_value(text) +def _trace_string(value: object, name: str) -> str: + return validate_value(value, name) def build_trace_query( @@ -252,53 +412,83 @@ def build_trace_query( attributes: Iterable[tuple[str, str]] | None = None, attributes_not: Iterable[tuple[str, str]] | None = None, attributes_regex: Iterable[tuple[str, str]] | None = None, + critical_path: object | None = None, ) -> str: - """Build a deterministic Traces query in Monium's selector syntax.""" - project_value = validate_project(project) if operation is not None and span_name is not None: raise SelectorError("pass only one of operation and span-name") operation_value = operation if operation is not None else span_name - clauses: list[tuple[str, str, str]] = [("project", "==", project_value)] - for key, value in ( - ("cluster", cluster), - ("service", service), - ("trace.id", trace_id), - ("span.id", span_id), - ("span.name", operation_value), - ("span.kind", span_kind), - ("span.status", span_status), + if error_only and span_status is not None and str(span_status).upper() != "ERROR": + raise SelectorError("--error-only conflicts with a non-ERROR --span-status") + clauses: list[SelectorClause] = [_clause("project", project_value, "==")] + # Trace semantic positions are strings. This prevents IDs such as 123, + # booleans, and duration-looking values from becoming unquoted literals. + for key, value, operator in ( + ("cluster", cluster, "=="), + ("service", service, "=="), + ("trace.id", trace_id, "=="), + ("span.id", span_id, "="), + ("span.name", operation_value, "=="), + ("span.kind", span_kind, "=="), + ("span.status", span_status, "="), ): if value is not None and value != "": - clauses.append((key, "==", validate_value(value, key))) + clauses.append(_clause(key, _trace_string(value, key), operator)) if error_only and span_status is None: - clauses.append(("span.status", "==", "ERROR")) + clauses.append(_clause("span.status", "ERROR", "=")) if min_duration is not None: - clauses.append(("span.duration", ">=", validate_duration(min_duration, "min-duration"))) + clauses.append(_clause("span.duration", validate_duration(min_duration, "min-duration"), ">=", value_type="duration")) if max_duration is not None: - clauses.append(("span.duration", "<=", validate_duration(max_duration, "max-duration"))) + clauses.append(_clause("span.duration", validate_duration(max_duration, "max-duration"), "<=", value_type="duration")) + if critical_path: + state = "PRESENT" if critical_path is True else validate_value(critical_path, "critical-path", max_length=16).upper() + if state not in {"PRESENT", "ABSENT"}: + raise SelectorError("critical-path must be PRESENT or ABSENT") + clauses.append(_clause("span.critical_path", state, "=")) for key, value in attributes or (): - clauses.append((validate_key(key, "attribute key"), "==", validate_value(value, "attribute value"))) + clauses.append( + _clause( + _validate_generic_field(key, family="attribute", reserved=_TRACE_RESERVED_ATTRIBUTE_FIELDS), + validate_value(value, "attribute value"), + "=", + ) + ) for key, value in attributes_not or (): - clauses.append((validate_key(key, "attribute key"), "!=", validate_value(value, "attribute value"))) + clauses.append( + _clause( + _validate_generic_field(key, family="attribute-not", reserved=_TRACE_RESERVED_ATTRIBUTE_FIELDS), + validate_value(value, "attribute value"), + "!=", + ) + ) for key, value in attributes_regex or (): - clauses.append((validate_key(key, "attribute key"), "=~", validate_value(value, "attribute value"))) - - rendered = [f"{key}{operator}{_trace_literal(value)}" for key, operator, value in clauses] - return "{ " + ", ".join(rendered) + " }" + clauses.append( + _clause( + _validate_generic_field(key, family="attribute-regex", reserved=_TRACE_RESERVED_ATTRIBUTE_FIELDS), + validate_value(value, "attribute value"), + "=~", + value_type="regex", + ) + ) + return _render_clauses(clauses) __all__ = [ + "ALLOWED_OPERATORS", "DEFAULT_LOG_COLUMNS", "LOG_TABS", + "SelectorClause", "SelectorError", "build_log_selector", "build_logs_url", "build_trace_query", + "parse_clause", "parse_key_value", "quote_log_value", + "validate_duration", + "validate_https_origin", "validate_key", + "validate_number", "validate_project", - "validate_duration", "validate_value", ] diff --git a/tests/test_config_redaction.py b/tests/test_config_redaction.py new file mode 100644 index 0000000..a8c01b1 --- /dev/null +++ b/tests/test_config_redaction.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from contextlib import redirect_stdout +from io import StringIO +from pathlib import Path + +from monium_cli.cli import main +from monium_cli.config import ConfigError, resolve_config + + +class ConfigErrorRedactionTests(unittest.TestCase): + """Config failures stay bounded and never echo user-controlled paths.""" + + CANARY = "MONIUM_CONFIG_CANARY_BEARER_TOKEN_7f9e" + + def _path(self, directory: str, suffix: str = "toml") -> Path: + return Path(directory) / f"monium-{self.CANARY}.{suffix}" + + def _assert_direct_error(self, path: Path, code: str) -> ConfigError: + with self.assertRaises(ConfigError) as raised: + resolve_config(config_path=path) + error = raised.exception + self.assertEqual(code, error.code) + self.assertNotIn(self.CANARY, str(error)) + self.assertNotIn(str(path), str(error)) + return error + + def _assert_cli_error(self, path: Path, code: str) -> dict[str, object]: + output = StringIO() + with redirect_stdout(output): + exit_code = main(["--config", str(path), "logs", "query", "--project", "scope"]) + self.assertEqual(2, exit_code) + raw = output.getvalue() + self.assertNotIn(self.CANARY, raw) + self.assertNotIn(str(path), raw) + payload = json.loads(raw) + self.assertFalse(payload["ok"]) + self.assertEqual(code, payload["error_code"]) + self.assertIsInstance(payload["error"], str) + self.assertNotIn(self.CANARY, payload["error"]) + self.assertNotIn(str(path), payload["error"]) + return payload + + def test_missing_path_is_redacted_in_exception_and_cli_json(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = self._path(directory) + self._assert_direct_error(path, "config_missing") + self._assert_cli_error(path, "config_missing") + + def test_directory_path_uses_bounded_unreadable_error(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = self._path(directory, "directory") + path.mkdir() + self._assert_direct_error(path, "config_unreadable") + self._assert_cli_error(path, "config_unreadable") + + def test_invalid_utf8_is_redacted(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = self._path(directory) + path.write_bytes(b"project = \"ok\"\n\xff\n") + self._assert_direct_error(path, "config_invalid_utf8") + self._assert_cli_error(path, "config_invalid_utf8") + + def test_invalid_toml_is_redacted(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = self._path(directory) + path.write_text("project = \"unterminated\n", encoding="utf-8") + self._assert_direct_error(path, "config_invalid_toml") + self._assert_cli_error(path, "config_invalid_toml") + + def test_unknown_key_does_not_echo_key_or_credential_value(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = self._path(directory) + path.write_text( + f'"{self.CANARY}" = "Bearer {self.CANARY}"\n', + encoding="utf-8", + ) + self._assert_direct_error(path, "config_unknown_key") + self._assert_cli_error(path, "config_unknown_key") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_monium_cli.py b/tests/test_monium_cli.py index 1dc14ea..2e3bd7b 100644 --- a/tests/test_monium_cli.py +++ b/tests/test_monium_cli.py @@ -3,6 +3,7 @@ import json import tempfile import unittest +import importlib.util from contextlib import redirect_stdout from io import StringIO from pathlib import Path @@ -22,13 +23,13 @@ class SelectorTests(unittest.TestCase): def test_logs_selector_uses_current_exact_and_contains_operators(self) -> None: query = build_log_selector( "folder__demo", - "payments", + "telemetry-api", cluster="production", level="error", contains='quote " safely', ) self.assertEqual( - '{ project=="folder__demo", cluster=="production", service=="payments", level==ERROR, message=*"quote \\\" safely" }', + '{ project=="folder__demo", cluster=="production", service=="telemetry-api", level==ERROR, message=*"quote \\\" safely" }', query, ) @@ -56,7 +57,7 @@ def test_trace_query_includes_ids_and_operation(self) -> None: min_duration="1s", ) self.assertEqual( - '{ project=="folder__demo", service=="api", trace.id=="trace-1", span.id=="span-1", span.name=="GET /health", span.duration>=1s }', + '{ project=="folder__demo", service=="api", trace.id=="trace-1", span.id="span-1", span.name=="GET /health", span.duration>=1s }', query, ) @@ -84,7 +85,7 @@ def test_common_monium_and_otel_shapes_are_normalized(self) -> None: "service": "api", "message": "failed", "trace_id": "t1", - "meta": [{"name": "RequestPath", "value": "/health"}], + "meta": [{"name": "route", "value": "/ready"}], } ), json.dumps( @@ -108,7 +109,7 @@ def test_common_monium_and_otel_shapes_are_normalized(self) -> None: self.assertEqual(2, result["matched"]) self.assertEqual("api", result["rows"][0]["service"]) self.assertEqual("worker", result["rows"][1]["service"]) - self.assertEqual("/health", result["rows"][0]["attributes"]["RequestPath"]) + self.assertEqual("/ready", result["rows"][0]["attributes"]["route"]) def test_nested_otel_envelope_and_limit_summary(self) -> None: path = self._write( @@ -148,14 +149,28 @@ def _run(self, argv: list[str]) -> tuple[int, dict]: return code, json.loads(output.getvalue()) def test_all_commands_emit_json(self) -> None: + code, payload = self._run(["capabilities"]) + self.assertEqual(0, code) + self.assertTrue(payload["ok"]) + self.assertEqual("offline", payload["mode"]) + by_name = {item["name"]: item for item in payload["capabilities"]} + self.assertTrue(by_name["logs.selector"]["available"]) + for name in ("alerts.list", "alerts.read", "logs.query.live"): + self.assertFalse(by_name[name]["available"]) + self.assertEqual("no_supported_provider", by_name[name]["reason_code"]) + self.assertFalse(payload["network_probed"]) + self.assertFalse(payload["credentials_probed"]) + code, payload = self._run(["logs", "url", "--project", "folder__x", "--service", "api"]) self.assertEqual(0, code) self.assertTrue(payload["ok"]) + self.assertEqual("offline", payload["mode"]) self.assertIn("url", payload) code, payload = self._run(["traces", "query", "--project", "folder__x"]) self.assertEqual(0, code) self.assertTrue(payload["ok"]) + self.assertEqual("offline", payload["mode"]) def test_open_is_only_called_when_requested(self) -> None: with patch("monium_cli.cli.webbrowser.open", return_value=True) as opener: @@ -180,6 +195,32 @@ def test_invalid_export_is_json_error_and_nonzero(self) -> None: self.assertFalse(payload["ok"]) self.assertIn("line 1", payload["error"]) + def test_argument_errors_do_not_echo_credentials(self) -> None: + canary = "password=super-secret-argv-canary" + code, payload = self._run(["--token", canary]) + self.assertNotEqual(0, code) + rendered = json.dumps(payload) + self.assertNotIn(canary, rendered) + self.assertNotIn("super-secret-argv-canary", rendered) + self.assertEqual("invalid_arguments", payload["error_code"]) + + def test_live_provider_surface_is_not_packaged_or_registered(self) -> None: + self.assertIsNone(importlib.util.find_spec("monium_cli.readonly")) + canary = "Bearer live-secret-canary" + with patch("monium_cli.cli.webbrowser.open") as opener: + attempts = ( + ["alerts", "list", "--project", "p", "--endpoint", canary], + ["alerts", "read", "--project", "p", "--id", "a", "--endpoint", canary], + ["logs", "query", "--live", "--token", canary], + ) + for argv in attempts: + with self.subTest(argv=argv): + code, payload = self._run(argv) + self.assertEqual(2, code) + self.assertEqual("invalid_arguments", payload["error_code"]) + self.assertNotIn("live-secret-canary", json.dumps(payload)) + opener.assert_not_called() + if __name__ == "__main__": unittest.main() diff --git a/tests/test_parser_output_bounds.py b/tests/test_parser_output_bounds.py new file mode 100644 index 0000000..01225c1 --- /dev/null +++ b/tests/test_parser_output_bounds.py @@ -0,0 +1,232 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from contextlib import redirect_stdout +from io import StringIO +from pathlib import Path +from unittest.mock import patch + +from monium_cli import parser +from monium_cli.cli import _emit, main +from monium_cli.parser import ExportError, parse_export + + +class ParserOutputBoundsTests(unittest.TestCase): + def _write(self, name: str, value: str | bytes) -> Path: + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + path = Path(directory.name) / name + if isinstance(value, bytes): + path.write_bytes(value) + else: + path.write_text(value, encoding="utf-8") + return path + + def _size(self, value: object) -> tuple[int, int]: + return parser._json_size(value) + + def test_unknown_field_overflow_fails_closed_including_summary_only(self) -> None: + record = {"message": "ok", **{f"unknown_{index}": index for index in range(2_001)}} + path = self._write("too-many-fields.ndjson", json.dumps(record) + "\n") + for summary_only in (False, True): + with self.subTest(summary_only=summary_only), self.assertRaises(ExportError) as raised: + parse_export(path, input_format="ndjson", summary_only=summary_only) + self.assertEqual("too_many_fields", raised.exception.code) + + def test_shared_resource_attributes_are_bounded_without_losing_summary(self) -> None: + export = { + "resource": {"attributes": {"shared": "x" * 500}}, + "scopeLogs": [{"logRecords": [{"message": "one"}, {"message": "two"}]}], + } + path = self._write("shared-resource.ndjson", json.dumps(export) + "\n") + with patch.object(parser, "MAX_OUTPUT_BYTES", 400): + result = parse_export(path, input_format="ndjson", include_attributes=True) + + output_bytes, _ = self._size(result) + self.assertLessEqual(output_bytes, 400) + self.assertEqual(2, result["matched"]) + self.assertEqual(0, result["returned"]) + self.assertTrue(result["rows_omitted"]) + self.assertTrue(any("output" in warning and "budget" in warning for warning in result["warnings"])) + + def test_summary_only_avoids_row_amplification(self) -> None: + export = { + "resource": {"attributes": {"shared": "x" * 500}}, + "scopeLogs": [{"logRecords": [{"message": "one"}, {"message": "two"}]}], + } + path = self._write("shared-summary.ndjson", json.dumps(export) + "\n") + with patch.object(parser, "MAX_OUTPUT_BYTES", 400): + result = parse_export(path, input_format="ndjson", include_attributes=True, summary_only=True) + + output_bytes, _ = self._size(result) + self.assertLessEqual(output_bytes, 400) + self.assertEqual(2, result["matched"]) + self.assertEqual([], result["rows"]) + self.assertTrue(result["rows_omitted"]) + self.assertFalse(any("output" in warning for warning in result["warnings"])) + + def test_output_byte_boundary_and_cap_plus_one(self) -> None: + path = self._write("boundary.ndjson", '{"message":"ok"}\n') + baseline = parse_export(path, input_format="ndjson") + baseline_bytes, _ = self._size(baseline) + + for cap in (baseline_bytes, baseline_bytes + 1): + with self.subTest(cap=cap), patch.object(parser, "MAX_OUTPUT_BYTES", cap): + result = parse_export(path, input_format="ndjson") + output_bytes, _ = self._size(result) + self.assertLessEqual(output_bytes, cap) + self.assertEqual(1, result["returned"]) + + # One byte below the full row is still safe because the parser makes + # the loss explicit and retains matching/summary counts. + with patch.object(parser, "MAX_OUTPUT_BYTES", baseline_bytes - 1): + result = parse_export(path, input_format="ndjson") + output_bytes, _ = self._size(result) + self.assertLessEqual(output_bytes, baseline_bytes - 1) + self.assertEqual(1, result["matched"]) + self.assertEqual(0, result["returned"]) + self.assertTrue(result["warnings"]) + + def test_output_node_boundary_and_cap_plus_one(self) -> None: + path = self._write("node-boundary.ndjson", '{"message":"ok"}\n') + baseline = parse_export(path, input_format="ndjson") + _, baseline_nodes = self._size(baseline) + + for cap in (baseline_nodes, baseline_nodes + 1): + with self.subTest(cap=cap), patch.object(parser, "MAX_OUTPUT_NODES", cap): + result = parse_export(path, input_format="ndjson") + _, output_nodes = self._size(result) + self.assertLessEqual(output_nodes, cap) + self.assertEqual(1, result["returned"]) + + with patch.object(parser, "MAX_OUTPUT_NODES", baseline_nodes - 1): + result = parse_export(path, input_format="ndjson") + _, output_nodes = self._size(result) + self.assertLessEqual(output_nodes, baseline_nodes - 1) + self.assertEqual(1, result["matched"]) + self.assertEqual(0, result["returned"]) + self.assertTrue(result["warnings"]) + + def test_cumulative_byte_budget_admits_partial_rows_only(self) -> None: + path = self._write( + "many-rows.ndjson", + "".join(json.dumps({"message": "x" * 100}) + "\n" for _ in range(20)), + ) + with patch.object(parser, "MAX_OUTPUT_BYTES", 1_000): + result = parse_export(path, input_format="ndjson", limit=20) + + output_bytes, _ = self._size(result) + self.assertLessEqual(output_bytes, 1_000) + self.assertEqual(20, result["matched"]) + self.assertEqual(2, result["returned"]) + self.assertTrue(result["truncated"]) + self.assertTrue(result["rows_omitted"]) + self.assertTrue(any("byte budget" in warning for warning in result["warnings"])) + + def test_cumulative_node_budget_admits_partial_rows_only(self) -> None: + path = self._write( + "many-nodes.ndjson", + "".join(json.dumps({"message": "x"}) + "\n" for _ in range(20)), + ) + with patch.object(parser, "MAX_OUTPUT_NODES", 100): + result = parse_export(path, input_format="ndjson", limit=20) + + _, output_nodes = self._size(result) + self.assertLessEqual(output_nodes, 100) + self.assertEqual(20, result["matched"]) + self.assertEqual(2, result["returned"]) + self.assertTrue(result["truncated"]) + self.assertTrue(result["rows_omitted"]) + self.assertTrue(any("JSON node budget" in warning for warning in result["warnings"])) + + def test_unrepresentable_minimum_budget_is_structured(self) -> None: + path = self._write("minimum.ndjson", '{"message":"ok"}\n') + with patch.object(parser, "MAX_OUTPUT_BYTES", 1), self.assertRaises(ExportError) as raised: + parse_export(path, input_format="ndjson") + self.assertEqual("output_too_large", raised.exception.code) + + with patch.object(parser, "MAX_OUTPUT_NODES", 1), self.assertRaises(ExportError) as raised: + parse_export(path, input_format="ndjson") + self.assertEqual("output_too_complex", raised.exception.code) + + def test_unreadable_path_is_not_echoed_in_error(self) -> None: + canary = "export-path-secret-canary" + missing = Path(tempfile.gettempdir()) / f"{canary}.ndjson" + with self.assertRaises(ExportError) as raised: + parse_export(missing, input_format="ndjson") + self.assertEqual("input_unreadable", raised.exception.code) + self.assertNotIn(canary, str(raised.exception)) + + def test_cli_emit_byte_boundary_includes_pretty_print_and_newline(self) -> None: + payload = { + "ok": True, + "schema_version": "1", + "tool_version": "0.2.0rc1", + "warnings": [], + "rows": [ + { + "fields": {"nested": {"items": [{"value": "x" * 32}, {"value": "y" * 32}]}}, + "message": "ready", + } + for _ in range(8) + ], + } + rendered = json.dumps(payload, ensure_ascii=True, allow_nan=False, indent=2, sort_keys=True) + wire_bytes = len(rendered.encode("ascii")) + 1 # print() newline + + with patch.object(parser, "MAX_OUTPUT_BYTES", wire_bytes): + output = StringIO() + with redirect_stdout(output): + emitted = _emit(payload) + self.assertTrue(emitted) + self.assertEqual(wire_bytes, len(output.getvalue().encode("ascii"))) + self.assertEqual(payload, json.loads(output.getvalue())) + + with patch.object(parser, "MAX_OUTPUT_BYTES", wire_bytes - 1): + output = StringIO() + with redirect_stdout(output): + emitted = _emit(payload) + self.assertFalse(emitted) + self.assertLessEqual(len(output.getvalue().encode("ascii")), wire_bytes - 1) + self.assertEqual("output_too_large", json.loads(output.getvalue())["error_code"]) + + def test_cli_emit_node_boundary_includes_envelope_nodes(self) -> None: + payload = { + "ok": True, + "schema_version": "1", + "tool_version": "0.2.0rc1", + "warnings": [], + "rows": [{"index": index, "nested": {"value": index}} for index in range(24)], + } + node_count = parser._json_node_count(payload) + + with patch.object(parser, "MAX_OUTPUT_NODES", node_count): + output = StringIO() + with redirect_stdout(output): + emitted = _emit(payload) + self.assertTrue(emitted) + self.assertEqual(payload, json.loads(output.getvalue())) + + with patch.object(parser, "MAX_OUTPUT_NODES", node_count - 1): + output = StringIO() + with redirect_stdout(output): + emitted = _emit(payload) + self.assertFalse(emitted) + fallback = json.loads(output.getvalue()) + self.assertEqual("output_too_complex", fallback["error_code"]) + self.assertLessEqual(parser._json_node_count(fallback), node_count - 1) + + def test_main_returns_nonzero_when_envelope_exceeds_cap(self) -> None: + output = StringIO() + with patch.object(parser, "MAX_OUTPUT_BYTES", 300), redirect_stdout(output): + code = main(["capabilities"]) + self.assertEqual(2, code) + payload = json.loads(output.getvalue()) + self.assertEqual("output_too_large", payload["error_code"]) + self.assertLessEqual(len(output.getvalue().encode("ascii")), 300) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_scope_overrides.py b/tests/test_scope_overrides.py new file mode 100644 index 0000000..d886055 --- /dev/null +++ b/tests/test_scope_overrides.py @@ -0,0 +1,149 @@ +from __future__ import annotations + +import json +import re +import unittest +from contextlib import redirect_stdout +from io import StringIO + +from monium_cli.cli import main +from monium_cli.selectors import SelectorClause, SelectorError, build_log_selector, build_trace_query + + +LOG_RESERVED = ("project", "cluster", "service", "log_group_id", "trace.id", "span.id") +TRACE_RESERVED = ( + "project", + "cluster", + "service", + "trace.id", + "span.id", + "span.name", + "span.kind", + "span.status", + "span.duration", + "span.critical_path", +) + + +class ScopeOverrideTests(unittest.TestCase): + def test_log_generic_filter_families_reject_reserved_scope_and_target_fields(self) -> None: + families = ( + ("exact", lambda key: build_log_selector("project", "service", fields=[(key, "value")])), + ("contains", lambda key: build_log_selector("project", "service", contains_fields=[(key, "value")])), + ("glob", lambda key: build_log_selector("project", "service", glob_fields=[(key, "value")])), + ("regex", lambda key: build_log_selector("project", "service", regex_fields=[(key, "value")])), + ( + "numeric", + lambda key: build_log_selector("project", "service", numeric_fields=[(key, "=", "1")]), + ), + ( + "clause", + lambda key: build_log_selector( + "project", "service", clauses=[SelectorClause(key, "==", "value")] + ), + ), + ) + for family, build in families: + for key in LOG_RESERVED: + with self.subTest(family=family, key=key): + error_pattern = rf"^generic filters cannot target first-class selector field: {re.escape(key)}$" + with self.assertRaisesRegex( + SelectorError, + error_pattern, + ): + build(key) + + def test_trace_generic_attribute_families_reject_reserved_selector_fields(self) -> None: + families = ( + ("attribute", lambda key: build_trace_query("project", attributes=[(key, "value")])), + ("attribute-not", lambda key: build_trace_query("project", attributes_not=[(key, "value")])), + ( + "attribute-regex", + lambda key: build_trace_query("project", attributes_regex=[(key, "value")]), + ), + ) + for family, build in families: + for key in TRACE_RESERVED: + with self.subTest(family=family, key=key): + error_pattern = rf"^generic filters cannot target first-class selector field: {re.escape(key)}$" + with self.assertRaisesRegex( + SelectorError, + error_pattern, + ): + build(key) + + def test_legitimate_generic_fields_and_metadata_remain_supported(self) -> None: + logs = build_log_selector( + "project", + "service", + fields=[("http.status_code", "200"), ("message", "ready")], + contains_fields=[("message", "read"), ("meta.route", "/health")], + glob_fields=[("meta.host", "web*")], + regex_fields=[("http.route", r"^/health")], + numeric_fields=[("latency_ms", ">=", "10")], + ) + self.assertIn('http.status_code=="200"', logs) + self.assertIn('message=="ready"', logs) + self.assertIn('message=*"read"', logs) + self.assertIn('meta.route=*"/health"', logs) + self.assertIn('meta.host="web*"', logs) + self.assertIn('http.route=~"^/health"', logs) + self.assertIn("latency_ms>=10", logs) + + traces = build_trace_query( + "project", + attributes=[("http.route", "/health")], + attributes_not=[("meta.region", "test")], + attributes_regex=[("http.method", "GET|HEAD")], + ) + self.assertIn('http.route="/health"', traces) + self.assertIn('meta.region!="test"', traces) + self.assertIn('http.method=~"GET|HEAD"', traces) + + def test_cli_rejects_reserved_fields_for_logs_and_traces(self) -> None: + log_attempts = ( + ("--field", "project=value"), + ("--contains-field", "service=value"), + ("--glob-field", "trace.id=value"), + ("--regex-field", "log_group_id=value"), + ("--number-field", "span.id=1"), + ) + for option, value in log_attempts: + with self.subTest(option=option): + code, payload = self._run( + ["logs", "query", "--project", "project", "--service", "service", option, value] + ) + self.assertEqual(2, code) + self.assertEqual("cli_error", payload["error_code"]) + self.assertEqual( + "generic filters cannot target first-class selector field: " + + value.split("=", 1)[0], + payload["error"], + ) + + trace_attempts = ( + ("--attr", "project=value"), + ("--attr-not", "span.status=value"), + ("--attr-regex", "span.duration=value"), + ) + for option, value in trace_attempts: + with self.subTest(option=option): + code, payload = self._run(["traces", "query", "--project", "project", option, value]) + self.assertEqual(2, code) + self.assertEqual("cli_error", payload["error_code"]) + self.assertEqual( + "generic filters cannot target first-class selector field: " + + value.split("=", 1)[0], + payload["error"], + ) + + @staticmethod + def _run(argv: list[str]) -> tuple[int, dict[str, object]]: + output = StringIO() + with redirect_stdout(output): + code = main(argv) + return code, json.loads(output.getvalue()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_v02_requirements.py b/tests/test_v02_requirements.py new file mode 100644 index 0000000..6c81a18 --- /dev/null +++ b/tests/test_v02_requirements.py @@ -0,0 +1,366 @@ +from __future__ import annotations + +import io +import json +import csv +import os +import subprocess +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest.mock import patch + +from monium_cli import __version__ +from monium_cli.cli import main +from monium_cli.config import ConfigError, resolve_config +from monium_cli.parser import ExportError, _BoundedCSVLines, parse_export +from monium_cli.selectors import ( + SelectorError, + build_log_selector, + build_logs_url, + build_trace_query, + validate_duration, + validate_https_origin, +) + + +class SelectorRequirementsTests(unittest.TestCase): + def test_typed_log_operator_golden_and_literal_star(self) -> None: + query = build_log_selector( + "scope", + "telemetry", + fields=[("env", "prod*")], + glob_fields=[("host", "web*")], + contains="literal * text", + contains_fields=[("meta.route", "/ready")], + ) + self.assertIn('env=="prod*"', query) + self.assertIn('host="web*"', query) + self.assertIn('message=*"literal * text"', query) + self.assertIn('meta.route=*"/ready"', query) + + def test_duration_boundaries(self) -> None: + for value in ("150us", "1ms", "2s", "3m", "4h", ".5s"): + with self.subTest(value=value): + self.assertEqual(value, validate_duration(value)) + for value in ("1ns", "1µs", "1d", "-1s", "fast"): + with self.subTest(value=value): + with self.assertRaises(SelectorError): + validate_duration(value) + + def test_critical_path_states_and_trace_string_literals(self) -> None: + for state in ("PRESENT", "ABSENT"): + query = build_trace_query("scope", span_id="123", span_status="true", operation="1s", critical_path=state) + self.assertIn('span.id="123"', query) + self.assertIn('span.status="true"', query) + self.assertIn('span.name=="1s"', query) + self.assertIn(f'span.critical_path="{state}"', query) + self.assertIn('span.id=', query) + self.assertNotIn("span.id==", query) + with self.assertRaises(SelectorError): + build_trace_query("scope", critical_path="unknown") + + def test_log_scope_alternatives_and_conflicts(self) -> None: + self.assertEqual('{ log_group_id=="group" }', build_log_selector(None, None, log_group_id="group")) + self.assertEqual('{ trace.id=="trace" }', build_log_selector(None, None, trace_id="trace")) + self.assertIn('project=="scope"', build_log_selector("scope", "telemetry")) + with self.assertRaises(SelectorError): + build_log_selector(None, "telemetry") + with self.assertRaises(SelectorError): + build_log_selector("scope", "telemetry", level="INFO", min_level="WARN") + with self.assertRaises(SelectorError): + build_trace_query("scope", error_only=True, span_status="WARN") + + def test_contains_restriction_and_removed_shortcuts(self) -> None: + with self.assertRaises(SelectorError): + build_log_selector("scope", "telemetry", contains_fields=[("service", "x")]) + with self.assertRaises(SelectorError): + build_log_selector("scope", "telemetry", contains_fields=[("meta.", "x")]) + with self.assertRaises(TypeError): + build_log_selector("scope", "telemetry", path="/removed") # type: ignore[call-arg] + + def test_https_origin_adversarial_cases(self) -> None: + self.assertEqual("https://example.test", validate_https_origin("https://example.test/")) + bad = ( + "http://example.test", + "https://user:pass@example.test", + "https://example.test/path", + "https://example.test?x=1", + "https://example.test#fragment", + "https://example.test:0", + "https://example.test:65536", + "https://example.test:", + "https://example.test/ leading", + "https://example.test\n", + ) + for value in bad: + with self.subTest(value=value): + with self.assertRaises(SelectorError): + validate_https_origin(value) + with self.assertRaises(SelectorError): + build_logs_url("scope", '{ project=="scope" }', base_url="https://example.test/path") + + +class ConfigRequirementsTests(unittest.TestCase): + def test_precedence_cli_env_toml_and_no_config(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "config.toml" + path.write_text('project = "toml"\nservice = "toml-service"\ncluster = "toml-cluster"\n', encoding="utf-8") + env = {"MONIUM_PROJECT": "env", "MONIUM_SERVICE": "env-service", "MONIUM_IGNORED": "x"} + resolved = resolve_config(config_path=path, environ=env, cli_values={"project": "cli"}) + self.assertEqual("cli", resolved.values["project"]) + self.assertEqual("env-service", resolved.values["service"]) + self.assertEqual("toml-cluster", resolved.values["cluster"]) + without_file = resolve_config(config_path=path, no_config=True, environ={}) + self.assertEqual({}, without_file.values) + + def test_unknown_toml_keys_and_credential_like_keys_fail(self) -> None: + with tempfile.TemporaryDirectory() as directory: + for key in ("token", "cookie", "credentials"): + path = Path(directory) / f"{key}.toml" + path.write_text(f'{key} = "secret"\n', encoding="utf-8") + with self.subTest(key=key), self.assertRaises(ConfigError): + resolve_config(config_path=path) + + +class ParserRequirementsTests(unittest.TestCase): + def _write(self, name: str, data: str | bytes) -> Path: + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + path = Path(directory.name) / name + if isinstance(data, bytes): + path.write_bytes(data) + else: + path.write_text(data, encoding="utf-8") + return path + + def test_all_explicit_formats_and_auto(self) -> None: + ndjson = self._write("records.ndjson", '{"message":"ndjson"}\n') + json_path = self._write("records.json", '[{"message":"json"}]') + csv_path = self._write("records.csv", "message,service\nhello,telemetry\n") + text_path = self._write("records.txt", "plain text\n") + self.assertEqual("ndjson", parse_export(ndjson, input_format="auto")["input_format"]) + self.assertEqual("json", parse_export(json_path, input_format="auto")["input_format"]) + self.assertEqual("csv", parse_export(csv_path, input_format="auto")["input_format"]) + self.assertEqual("text", parse_export(text_path, input_format="auto")["input_format"]) + self.assertEqual("json", parse_export(json_path, input_format="json")["input_format"]) + self.assertEqual("telemetry", parse_export(csv_path, input_format="csv")["rows"][0]["service"]) + self.assertEqual("plain text", parse_export(text_path, input_format="text")["rows"][0]["message"]) + + def test_stdin_requires_format_and_accepts_csv(self) -> None: + with patch("sys.stdin", io.StringIO("message,service\nhello,telemetry\n")): + result = parse_export("-", input_format="csv") + self.assertEqual("hello", result["rows"][0]["message"]) + with patch("sys.stdin", io.StringIO('{"message":"ambiguous"}\n')): + with self.assertRaisesRegex(ExportError, "requires --input-format"): + parse_export("-", input_format="auto") + + def test_csv_embedded_newline_and_malformed_error(self) -> None: + path = self._write("embedded.csv", 'message,service\n"hello\nworld",telemetry\n') + result = parse_export(path, input_format="csv") + self.assertEqual("hello\nworld", result["rows"][0]["message"]) + bad = self._write("bad.csv", 'message,service\n"unterminated,telemetry\n') + with self.assertRaisesRegex(ExportError, "invalid CSV"): + parse_export(bad, input_format="csv") + + def test_csv_provenance_is_record_start_and_state_is_o1(self) -> None: + prefix = "\n" * 10_000 + data = prefix + "message,service\n\n\n\"hello\nworld\",telemetry\n" + path = self._write("many-blank.csv", data) + before_limit = csv.field_size_limit() + result = parse_export(path, input_format="csv") + self.assertEqual(before_limit, csv.field_size_limit()) + row = result["rows"][0] + self.assertEqual(10_004, row["line"]) + self.assertEqual(len(prefix + "message,service\n\n\n"), row["source"]["byte"]) + + source = _BoundedCSVLines(io.BytesIO(data.encode("utf-8"))) + reader = csv.reader(source, strict=True) + source.begin_record() + while not next(reader): + source.begin_record() + source.begin_record() + while not next(reader): + source.begin_record() + self.assertFalse(hasattr(source, "line_starts")) + self.assertEqual(10_004, source.record_start_line) + + def test_csv_field_limit_is_locked_restored_and_width_is_strict(self) -> None: + path = self._write("limit.csv", "message,service\nhello,telemetry\n") + expected = parse_export(path, input_format="csv") + ambient = csv.field_size_limit() + for ambient_limit in (8, 1_000_000): + csv.field_size_limit(ambient_limit) + try: + self.assertEqual(expected["rows"], parse_export(path, input_format="csv")["rows"]) + self.assertEqual(ambient_limit, csv.field_size_limit()) + finally: + csv.field_size_limit(ambient) + + extra_header = self._write("extra-header.csv", "a,_extra\n1,2\n") + row = parse_export(extra_header, input_format="csv")["rows"][0] + self.assertEqual("2", row["fields"]["_extra"]) + for name, contents in (("missing.csv", "a,b\n1\n"), ("extra.csv", "a,b\n1,2,3\n")): + bad = self._write(name, contents) + with self.subTest(name=name), self.assertRaisesRegex(ExportError, "CSV row") as raised: + parse_export(bad, input_format="csv") + self.assertEqual("invalid_csv", raised.exception.code) + + for name, contents in ( + ("duplicate-header.csv", "message,message\nfirst,second\n"), + ("empty-header.csv", "message,\nhello,value\n"), + ): + bad = self._write(name, contents) + with self.subTest(name=name), self.assertRaises(ExportError) as raised: + parse_export(bad, input_format="csv") + self.assertEqual("invalid_csv", raised.exception.code) + + def test_generic_container_names_are_fields(self) -> None: + record = {"message": "outer", "data": "d", "payload": "p", "logs": ["l"], "records": ["r"], "items": ["i"], "value": "v"} + path = self._write("generic.ndjson", json.dumps(record) + "\n") + row = parse_export(path, input_format="ndjson")["rows"][0] + self.assertEqual("d", row["fields"]["data"]) + self.assertEqual(["r"], row["fields"]["records"]) + self.assertEqual("v", row["fields"]["value"]) + empty = self._write("empty.ndjson", '{"records":[]}\n') + self.assertEqual(0, parse_export(empty, input_format="ndjson")["total_scanned"]) + + def test_otlp_time_anyvalue_and_severity(self) -> None: + value = { + "resourceLogs": [{ + "resource": {"attributes": [{"key": "service.name", "value": {"stringValue": "telemetry"}}]}, + "scopeLogs": [{"logRecords": [{ + "timeUnixNano": "1720000000000000000", + "severityNumber": 17, + "body": {"stringValue": "ready"}, + "attributes": [{"key": "tags", "value": {"arrayValue": {"values": [{"stringValue": "a"}, {"intValue": 2}]}}}, {"key": "kv", "value": {"kvlistValue": {"values": [{"key": "ok", "value": {"boolValue": True}}]}}}], + }]}], + }], + } + path = self._write("otel.ndjson", json.dumps(value) + "\n") + row = parse_export(path, input_format="ndjson", include_attributes=True)["rows"][0] + self.assertEqual("ERROR", row["level"]) + self.assertTrue(row["time"].endswith("Z")) + self.assertEqual(["a", 2], row["attributes"]["tags"]) + self.assertEqual({"ok": True}, row["attributes"]["kv"]) + + def test_strict_json_errors_location_and_bounds(self) -> None: + for payload in ('{"value":NaN}\n', '{"value":Infinity}\n', '{"value":1e9999}\n', '{"value":' + "9" * 101 + '}\n'): + path = self._write("bad.ndjson", payload) + with self.subTest(payload=payload[:20]), self.assertRaises(ExportError): + parse_export(path, input_format="ndjson") + invalid_utf8 = self._write("utf8.ndjson", b'{"message":"ok"}\n\xff\n') + with self.assertRaisesRegex(ExportError, r"line 2 byte"): + parse_export(invalid_utf8, input_format="ndjson") + deep = self._write("deep.json", "{" * 110 + "null" + "}" * 110) + with self.assertRaises(ExportError): + parse_export(deep, input_format="json") + + def test_allocation_bounds_cardinality_and_summary_only(self) -> None: + path = self._write("large.ndjson", b"{" + b"x" * 64 + b"}\n") + with patch("monium_cli.parser.MAX_LINE_BYTES", 16), self.assertRaisesRegex(ExportError, "exceeds"): + parse_export(path, input_format="ndjson") + csv_path = self._write("large.csv", b"message\n" + b"x" * 64 + b"\n") + with patch("monium_cli.parser.MAX_LINE_BYTES", 16), self.assertRaisesRegex(ExportError, "exceeds"): + parse_export(csv_path, input_format="csv") + text_path = self._write("large.txt", b"x" * 64 + b"\n") + with patch("monium_cli.parser.MAX_LINE_BYTES", 16), self.assertRaisesRegex(ExportError, "exceeds"): + parse_export(text_path, input_format="text") + input_path = self._write("input.ndjson", b'{"message":"1"}\n{"message":"2"}\n') + with patch("monium_cli.parser.MAX_INPUT_BYTES", 20), self.assertRaisesRegex(ExportError, "input exceeds"): + parse_export(input_path, input_format="ndjson") + rows = "\n".join(json.dumps({"level": str(101 + i), "message": str(i)}) for i in range(4)) + "\n" + summary_path = self._write("summary.ndjson", rows) + with patch("monium_cli.parser.MAX_SUMMARY_CARDINALITY", 2): + result = parse_export(summary_path, input_format="ndjson", summary_only=True, limit=1) + self.assertTrue(result["rows_omitted"]) + self.assertEqual([], result["rows"]) + self.assertTrue(result["truncated"]) + self.assertTrue(result["warnings"]) + + def test_ascii_locale_machine_output_and_no_traceback(self) -> None: + env = dict(os.environ) + env.update({"LC_ALL": "C", "PYTHONPATH": str(Path(__file__).parents[1] / "src")}) + proc = subprocess.run( + [sys.executable, "-m", "monium_cli", "logs", "query", "--project", "scope", "--service", "telemetry"], + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + self.assertEqual(0, proc.returncode, proc.stderr.decode("ascii", "replace")) + proc.stdout.decode("ascii") + payload = json.loads(proc.stdout) + self.assertEqual("1", payload["schema_version"]) + self.assertEqual(__version__, payload["tool_version"]) + bad = self._write("bad.ndjson", "not-json\n") + output = io.StringIO() + with redirect_stdout(output): + code = main(["logs", "parse", str(bad)]) + self.assertEqual(2, code) + self.assertIn("error_code", json.loads(output.getvalue())) + + +class CliRequirementsTests(unittest.TestCase): + def _run(self, argv: list[str], *, env: dict[str, str] | None = None) -> tuple[int, dict[str, object]]: + output = io.StringIO() + with patch.dict(os.environ, env or {}, clear=False), redirect_stdout(output): + code = main(argv) + return code, json.loads(output.getvalue()) + + def test_query_scope_paths_and_critical_path_cli(self) -> None: + code, payload = self._run(["logs", "query", "--log-group-id", "group"]) + self.assertEqual(0, code) + self.assertNotIn("project==", payload["query"]) + code, payload = self._run(["logs", "query", "--trace-id", "trace"]) + self.assertEqual(0, code) + self.assertIn('trace.id=="trace"', payload["query"]) + code, payload = self._run(["logs", "url", "--service", "telemetry"]) + self.assertEqual(2, code) + self.assertEqual("missing_project", payload["error_code"]) + code, payload = self._run(["traces", "query", "--project", "scope", "--critical-path", "ABSENT"]) + self.assertEqual(0, code) + self.assertIn('span.critical_path="ABSENT"', payload["query"]) + code, payload = self._run(["logs", "query", "--project", "scope", "--service", "telemetry", "--number-field", "telemetry.status=503"]) + self.assertEqual(0, code) + self.assertIn("telemetry.status=503", payload["query"]) + code, payload = self._run(["logs", "query", "--project", "scope", "--service", "telemetry", "--number", "latency", "==", "5"]) + self.assertEqual(2, code) + self.assertEqual("cli_error", payload["error_code"]) + + def test_browser_opt_in_false_and_error(self) -> None: + with patch("monium_cli.cli.webbrowser.open", side_effect=RuntimeError("browser unavailable")) as opener: + code, payload = self._run(["logs", "url", "--project", "scope", "--service", "telemetry", "--open"]) + self.assertEqual(0, code) + opener.assert_called_once() + self.assertFalse(payload["opened"]) + self.assertIn("open_error", payload) + with patch("monium_cli.cli.webbrowser.open") as opener: + code, payload = self._run(["logs", "url", "--project", "scope", "--service", "telemetry"]) + self.assertEqual(0, code) + opener.assert_not_called() + self.assertFalse(payload["open_requested"]) + + def test_config_cli_over_env_and_toml(self) -> None: + with tempfile.TemporaryDirectory() as directory: + config = Path(directory) / "config.toml" + config.write_text('project = "toml"\nservice = "toml-service"\n', encoding="utf-8") + code, payload = self._run(["--config", str(config), "logs", "query", "--service", "cli"], env={"MONIUM_PROJECT": "env"}) + self.assertEqual(0, code) + self.assertIn('project=="env"', payload["query"]) + self.assertIn('service=="cli"', payload["query"]) + + def test_version_and_ci_source_layout_contract(self) -> None: + self.assertTrue(__version__.startswith("0.2.0rc1")) + root = Path(__file__).parents[1] + self.assertTrue((root / "src/monium_cli/config.py").is_file()) + self.assertTrue((root / ".github/dependabot.yml").is_file()) + workflow = (Path(__file__).parents[1] / ".github/workflows/ci.yml").read_text(encoding="utf-8") + self.assertIn("PYTHONPATH: src", workflow) + self.assertIn("contents: read", workflow) + + +if __name__ == "__main__": + unittest.main()