Release 0.2.7: default log redaction

README.md

@@ -1,5 +1,27 @@
 # TeaQL Python SDK
 
+## Sensitive log data
+
+Runtime diagnostic logs redact payload values by default, before delivery to
+file, console, buffers, or custom logging sinks. Selecting a diagnostic sink
+alone does not authorize plaintext. For controlled troubleshooting only:
+
+```bash
+export TEAQL_ALLOW_SENSITIVE_PLAINTEXT_LOGS=I_UNDERSTAND_SENSITIVE_DATA_MAY_BE_WRITTEN_TO_DISK
+```
+
+Only this exact value enables plaintext permission; empty values, `true`, and
+whitespace variants do not. Enabling it emits a warning. Credential-classified
+fields remain redacted. The flag does not force every sink to expose values.
+SQL without reliable field/literal provenance may be suppressed and marked
+`NOT REPLAYABLE`. Execution parameters and persisted business data are unchanged.
+
+Do not put sensitive data in free-text comments or purpose declarations.
+TeaQL cannot govern arbitrary application prints or independent driver loggers;
+configure those separately. This setting does not erase older plaintext files.
+Restrict access and retention when using plaintext diagnostics, then unset the
+variable and restart processes when troubleshooting is complete.
+
 TeaQL Python SDK is a runtime engine and toolkit for building data-driven business applications. It provides seamless integration with the TeaQL ecosystem, fully aligned with the `teaql-rs` baseline.
 
 ## Recommended Agent Harness

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "teaql"
-version = "0.2.6"
+version = "0.2.7"
 description = "TeaQL Runtime and Data Service for Python"
 readme = "README.md"
 requires-python = ">=3.10"

releases/0.2.7.md

@@ -0,0 +1,34 @@
+# 0.2.7: default log privacy
+
+Tracking: https://github.com/teaql/teaql-python/issues/25
+Cross-runtime contract: https://github.com/teaql/teaql-conformance/issues/58
+
+## Behavior and migration
+
+Runtime-controlled SQL and audit logging now redact sensitive payloads before
+normal, custom, buffered and file destinations. Execution values are unchanged.
+Selecting a diagnostic sink alone no longer authorizes plaintext output.
+
+For intentional plaintext debugging only, set this exact acknowledgement:
+
+```sh
+export TEAQL_ALLOW_SENSITIVE_PLAINTEXT_LOGS=I_UNDERSTAND_SENSITIVE_DATA_MAY_BE_WRITTEN_TO_DISK
+```
+
+Invalid, empty or truthy shortcut values do not enable it. Enabling it emits a
+value-free warning. Credential-classified fields remain hidden. The switch grants
+permission; it does not force every sink to reveal values. Restart after changing
+the environment. Previously written files are not erased.
+
+Without reliable parameter/field provenance, SQL logs use conservative redaction;
+SQL may be labeled non-replayable. Operations, timing and safe intent remain.
+Do not put unrelated secrets in comments/purpose, or print raw business events.
+Arbitrary application and third-party driver logs are outside this boundary.
+
+## Verification
+
+Local runtime tests and `bash scripts/verify-examples.sh` passed before release
+preparation. Retained real SQLite CRUD/failure canaries assert unchanged persisted
+values and absence of sensitive markers in runtime log destinations. Exact/invalid
+environment gates and credential behavior have separate regression fixtures.
+Published-package verification is a separate gate, not implied by local results.

src/teaql/runtime/audit.py

@@ -29,19 +29,28 @@ class RawAuditEvent:
     category: Optional[str] = None
 
     def safe(self, mask_fields: List[str], max_length: Optional[int]) -> "SafeAuditEvent":
+        from .log_privacy import REDACTED, credential_name, payload_has_credentials, plaintext_enabled, scrub, value_strings
+        allow = plaintext_enabled()
+        secrets = []
         fields = []
         for change in self.changes:
             value = None if change.new_value is None else str(getattr(change.new_value, "val", change.new_value))
-            masked = change.field in mask_fields
+            masked = (credential_name(change.field)
+                      or payload_has_credentials(change.new_value)
+                      or payload_has_credentials(change.old_value)
+                      or (change.field in mask_fields and not allow))
+            if masked:
+                secrets.extend(value_strings(change.old_value))
+                secrets.extend(value_strings(change.new_value))
             if value is not None and masked:
-                value = _mask(value)
+                value = REDACTED
             truncated = value is not None and max_length is not None and len(value) > max_length
             if truncated:
                 value = "*" * max_length if max_length <= 3 else value[:max_length - 3] + "..."
             fields.append(SafeAuditField(change.field, value, masked, truncated))
         return SafeAuditEvent(
-            self.kind, self.entity, self.entity_id, tuple(fields), self.trace_chain,
-            self.actor, self.category,
+            self.kind, self.entity, self.entity_id, scrub(tuple(fields), secrets), scrub(self.trace_chain, secrets),
+            scrub(self.actor, secrets), self.category,
         )
 
 

src/teaql/runtime/context.py

@@ -739,6 +739,8 @@ def record_metadata_log(self, metadata: Any):
         elif entry.affected_rows is not None:
             entry.result_summary = f"{entry.affected_rows} rows affected"
 
+        from .log_privacy import sql_log_projection
+        entry = sql_log_projection(entry)
         logs = self.sql_logs()
         logs.append(entry)
         self._resources["sql_logs"] = logs
@@ -750,7 +752,7 @@ def record_metadata_log(self, metadata: Any):
             buf.entries.append(UnifiedLogEntry(
                 timestamp=entry.started_at,
                 user_identifier=self.user_identifier(),
-                trace_chain=getattr(metadata, 'trace_chain', []),
+                trace_chain=entry.trace_path,
                 payload=LogPayload.Sql(entry)
             ))
 
@@ -784,6 +786,8 @@ def record_sql_log(self, operation: Any, query: Any, started_at: Any, ended_at:
         elif affected_rows is not None:
             entry.result_summary = f"{affected_rows} rows affected"
 
+        from .log_privacy import sql_log_projection
+        entry = sql_log_projection(entry)
         logs = self.sql_logs()
         logs.append(entry)
         self._resources["sql_logs"] = logs
@@ -1033,7 +1037,7 @@ class SqlLogEntry:
     result_summary: str
 
 class DiagnosticSqlLogSink:
-    """Value-bearing diagnostic SQL destination; the text sink is installed by default."""
+    """Policy-projected SQL destination; the text sink is installed by default."""
     def write(self, entry: SqlLogEntry) -> None:
         raise NotImplementedError
 
@@ -1042,6 +1046,8 @@ def __init__(self, writer=print):
         self._writer = writer
 
     def write(self, entry: SqlLogEntry) -> None:
+        from .log_privacy import sql_log_projection
+        entry = sql_log_projection(entry)
         elapsed_us = int(entry.elapsed.total_seconds() * 1_000_000) if entry.elapsed else 0
         self._writer(
             f"[TeaQL SQL][{entry.operation.name.lower()}][{elapsed_us}us] "

src/teaql/runtime/log_privacy.py

@@ -0,0 +1,93 @@
+"""Log-only projections. Never modify execution parameters or stored values."""
+
+import logging
+import os
+import re
+from dataclasses import fields, is_dataclass, replace
+from functools import lru_cache
+
+PLAINTEXT_ENV = "TEAQL_ALLOW_SENSITIVE_PLAINTEXT_LOGS"
+PLAINTEXT_ACK = "I_UNDERSTAND_SENSITIVE_DATA_MAY_BE_WRITTEN_TO_DISK"
+REDACTED = "[REDACTED]"
+SQL_REDACTED = "[REDACTED SQL; NOT REPLAYABLE]"
+
+
+@lru_cache(maxsize=1)
+def _warn_plaintext():
+    logging.getLogger("teaql.runtime").warning(
+        "Sensitive plaintext logging enabled: application data may be written to disk. "
+        "Authentication secrets remain redacted."
+    )
+
+
+def plaintext_enabled():
+    enabled = os.environ.get(PLAINTEXT_ENV) == PLAINTEXT_ACK
+    if enabled:
+        _warn_plaintext()
+    return enabled
+
+
+def credential_name(name):
+    normalized = re.sub(r"[^a-z0-9]", "", str(name).lower())
+    return any(word in normalized for word in (
+        "password", "passwd", "passphrase", "privatekey", "secret", "accesstoken",
+        "refreshtoken", "idtoken", "apikey", "authorization", "credential",
+        "sessiontoken", "magiclinktoken",
+    ))
+
+
+def payload_has_credentials(value):
+    value = getattr(value, "val", value)
+    if isinstance(value, dict):
+        return any(credential_name(k) or payload_has_credentials(v) for k, v in value.items())
+    if isinstance(value, (tuple, list)):
+        return any(payload_has_credentials(v) for v in value)
+    return False
+
+
+def value_strings(value):
+    value = getattr(value, "val", value)
+    if isinstance(value, dict):
+        return [s for v in value.values() for s in value_strings(v)]
+    if isinstance(value, (tuple, list)):
+        return [s for v in value for s in value_strings(v)]
+    return [] if value is None or value == "" else [str(value)]
+
+
+def scrub(value, secrets):
+    """Copy annotations, removing known values even when embedded in intent."""
+    if isinstance(value, str):
+        for secret in sorted(set(secrets), key=len, reverse=True):
+            value = value.replace(secret, REDACTED)
+        return value
+    if isinstance(value, list):
+        return [scrub(v, secrets) for v in value]
+    if isinstance(value, tuple):
+        return tuple(scrub(v, secrets) for v in value)
+    if isinstance(value, dict):
+        return {k: scrub(v, secrets) for k, v in value.items()}
+    if is_dataclass(value):
+        return replace(value, **{f.name: scrub(getattr(value, f.name), secrets)
+                                 for f in fields(value) if f.init})
+    return value
+
+
+def sql_log_projection(entry):
+    # SQL has no per-parameter field provenance. A credential-bearing query
+    # therefore suppresses its whole payload, including on debug opt-in.
+    credentials = (credential_name(entry.sql) or credential_name(entry.debug_sql)
+                   or payload_has_credentials(entry.params))
+    allow = plaintext_enabled() and not credentials
+    if allow:
+        return replace(entry, params=list(entry.params), trace_path=list(entry.trace_path))
+    secrets = value_strings(entry.params)
+    sql = entry.sql
+    # Arbitrary literal SQL is not safely redactable across dialects. Preserve
+    # parameterized shape only when there are no literals/comments to expose.
+    if any(token in sql for token in ("'", '"', "`", "--", "/*", "$")) or re.search(r"\b\d+\b", sql):
+        sql = SQL_REDACTED
+    return replace(entry, sql=scrub(sql, secrets), params=[None] * len(entry.params),
+                   debug_sql=SQL_REDACTED, pretty_sql=SQL_REDACTED,
+                   comment=scrub(entry.comment, secrets), purpose=scrub(entry.purpose, secrets),
+                   audit_reason=scrub(entry.audit_reason, secrets),
+                   trace_path=scrub(entry.trace_path, secrets))

tests/core/test_context.py

@@ -38,8 +38,8 @@ class MockMetadata:
     context.set_diagnostic_sql_log_sink(TextDiagnosticSqlLogSink(output.append))
     context.record_metadata_log(MockMetadata())
     assert len(context.sql_logs()) == 1
-    assert context.sql_logs()[0].debug_sql == "SELECT 1"
-    assert output[0].endswith("\nDebug SQL: SELECT 1")
+    assert context.sql_logs()[0].debug_sql == "[REDACTED SQL; NOT REPLAYABLE]"
+    assert output[0].endswith("\nDebug SQL: [REDACTED SQL; NOT REPLAYABLE]")
     assert "Parameterized SQL:" in output[0]
 
     # SQL logs

tests/provider/sqlite/test_sqlite.py

@@ -236,6 +236,40 @@ async def test_crud(temp_db, schema_provider, service):
     query_res = await service.query(None, query_req)
     assert len(query_res.rows) == 0
 
+@pytest.mark.asyncio
+async def test_crud_privacy_preserves_sqlite_values_and_failure(temp_db, service, tmp_path, monkeypatch):
+    from teaql.runtime.context import TextDiagnosticSqlLogSink
+    from teaql.runtime.log_privacy import PLAINTEXT_ENV
+    monkeypatch.delenv(PLAINTEXT_ENV, raising=False)
+    async with aiosqlite.connect(temp_db) as db:
+        await db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, version INTEGER)")
+        await db.commit()
+    first, second, failed = "PRIVATE-CREATE-CANARY", "PRIVATE-UPDATE-CANARY", "PRIVATE-FAILURE-CANARY"
+    context = RuntimeModule.new().into_context()
+    path = tmp_path / "runtime.log"
+    with path.open("w") as output:
+        context.set_diagnostic_sql_log_sink(TextDiagnosticSqlLogSink(lambda line: print(line, file=output)))
+        def insert(name):
+            return MutationRequest(InsertCommand("User", {
+                "id": Value.I64(1), "name": Value.Text(name), "version": Value.I64(1)}))
+        created = await service.mutate(context, insert(first))
+        assert created.persisted_record["name"] == first
+        query = QueryRequest(SelectQuery("User").limit(1)).comment("read privacy fixture").purpose("verify original stored values")
+        assert (await service.query(context, query)).rows[0]["name"] == first
+        updated = await service.mutate(context, MutationRequest(
+            UpdateCommand("User", Value.I64(1)).value("name", Value.Text(second))))
+        assert updated.persisted_record["name"] == second
+        with pytest.raises(Exception):
+            await service.mutate(context, insert(failed))
+        assert (await service.query(context, query)).rows[0]["name"] == second
+        await service.mutate(context, MutationRequest(DeleteCommand("User", Value.I64(1)).hard_delete()))
+        assert (await service.query(context, query)).rows == []
+    logged = path.read_text() + repr(context.sql_logs())
+    for marker in (first, second, failed):
+        assert marker not in logged
+    for operation in ("insert", "select", "update", "delete"):
+        assert operation in logged
+
 @pytest.mark.asyncio
 async def test_id_set_pagination_jumps_restores_order_and_avoids_count(temp_db, schema_provider):
     service = create_sqlite_service(temp_db, schema_provider)

tests/runtime/test_log_privacy.py

@@ -0,0 +1,99 @@
+from dataclasses import replace
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+
+import pytest
+
+from teaql.runtime.audit import AuditFieldChange, MutationAuditKind, RawAuditEvent
+from teaql.runtime.context import UserContext, SqlLogEntry, SqlLogOperation, TextDiagnosticSqlLogSink
+from teaql.runtime.log_privacy import PLAINTEXT_ENV, PLAINTEXT_ACK, REDACTED, sql_log_projection
+
+
+@pytest.fixture(autouse=True)
+def safe_environment(monkeypatch):
+    monkeypatch.delenv(PLAINTEXT_ENV, raising=False)
+
+
+def entry(field="name", value="PRIVATE-CUSTOMER-CANARY"):
+    return SqlLogEntry(SqlLogOperation.Update, f"update {value}", "customer edit", None,
+                       [f"edit {value}"], f"UPDATE customer SET {field} = ?", [value],
+                       f"UPDATE customer SET {field} = '{value}'", "", datetime.now(),
+                       datetime.now(), timedelta(microseconds=1), None, None, 1, "1 rows affected")
+
+
+@pytest.mark.parametrize("setting", [None, "", "true", "1", PLAINTEXT_ACK + " ", PLAINTEXT_ACK.lower()])
+def test_default_and_invalid_ack_protect_file_and_copy(monkeypatch, tmp_path, setting):
+    if setting is not None:
+        monkeypatch.setenv(PLAINTEXT_ENV, setting)
+    raw = entry()
+    with (tmp_path / "runtime.log").open("w") as output:
+        TextDiagnosticSqlLogSink(lambda text: print(text, file=output)).write(raw)
+    logged = (tmp_path / "runtime.log").read_text()
+    assert raw.params[0] not in logged
+    assert "1 rows affected" in logged
+    assert "NOT REPLAYABLE" in logged
+    assert raw.params == ["PRIVATE-CUSTOMER-CANARY"]
+    assert "PRIVATE-CUSTOMER-CANARY" in raw.debug_sql
+
+
+def test_context_sanitizes_before_custom_sink_and_buffer():
+    raw = entry()
+    captured = []
+    context = UserContext.new()
+    context.set_diagnostic_sql_log_sink(SimpleNamespace(write=captured.append))
+    metadata = SimpleNamespace(operation="update", parameters=raw.params,
+                               parameterized_sql=raw.sql, debug_query=raw.debug_sql,
+                               comment=raw.comment, trace_chain=raw.trace_path, affected_rows=1)
+    context.record_metadata_log(metadata)
+    assert raw.params[0] not in repr(captured)
+    assert raw.params[0] not in repr(context.sql_logs())
+    assert metadata.parameters == raw.params
+
+
+def test_exact_opt_in_warns_and_preserves_noncredential_debug(monkeypatch, caplog):
+    from teaql.runtime.log_privacy import _warn_plaintext
+    _warn_plaintext.cache_clear()
+    monkeypatch.setenv(PLAINTEXT_ENV, PLAINTEXT_ACK)
+    raw = entry()
+    output = []
+    TextDiagnosticSqlLogSink(output.append).write(raw)
+    assert raw.params[0] in output[0]
+    assert "may be written to disk" in caplog.text
+    assert raw.params[0] not in caplog.text
+
+
+@pytest.mark.parametrize("field", ["password", "accessToken", "private_key", "api_key", "refresh_token"])
+def test_credentials_never_reach_sql_sink(monkeypatch, field):
+    monkeypatch.setenv(PLAINTEXT_ENV, PLAINTEXT_ACK)
+    raw = entry(field)
+    assert raw.params[0] not in repr(sql_log_projection(raw))
+
+
+@pytest.mark.parametrize("sql", ["SELECT 'UNIQUE-SECRET'", "SELECT 123456", "SELECT 1 -- UNIQUE-SECRET",
+                                 "SELECT $$UNIQUE-SECRET$$", 'SELECT "UNIQUE-SECRET"'])
+def test_unclassified_literal_sql_is_not_logged(sql):
+    safe = sql_log_projection(replace(entry(), sql=sql, params=[], debug_sql=sql))
+    assert safe.sql == "[REDACTED SQL; NOT REPLAYABLE]"
+    assert safe.debug_sql == safe.sql
+
+
+@pytest.mark.parametrize("enabled", [False, True])
+def test_audit_field_policy_and_credential_override(monkeypatch, enabled):
+    if enabled:
+        monkeypatch.setenv(PLAINTEXT_ENV, PLAINTEXT_ACK)
+    event = RawAuditEvent(MutationAuditKind.UPDATED, "Customer", 1, (
+        AuditFieldChange("email", "old-private-email", "new-private-email"),
+        AuditFieldChange("password", "old-password-secret", "new-password-secret"),
+    ), ("change new-password-secret",))
+    safe = event.safe(["email"], None)
+    assert safe.fields[0].value == ("new-private-email" if enabled else REDACTED)
+    assert safe.fields[1].value == REDACTED
+    assert "new-password-secret" not in repr(safe)
+    assert event.changes[1].new_value == "new-password-secret"
+
+
+def test_nested_credentials_stay_masked(monkeypatch):
+    monkeypatch.setenv(PLAINTEXT_ENV, PLAINTEXT_ACK)
+    event = RawAuditEvent(MutationAuditKind.CREATED, "Account", 1,
+                          (AuditFieldChange("settings", None, {"apiKey": "NESTED-SECRET"}),))
+    assert "NESTED-SECRET" not in repr(event.safe([], None))