feat: is_encrypted 加密状态判断与未加密写入通道 (0.2.0)

- is_encrypted(): certutil 原始哈希 + helper 视图对比, Python/pyd/EXE 三环境均可判定
- w/a/x 写入按后缀探测通道, 直写会被加密时自动切换 PowerShell stdin 中转, 保证落盘未加密
- 新增 WriteChannel/DlpWriteError 与 DLP_IO_WRITE_CHANNEL 强制通道开关
- tests: DLP 加解密样本 fixture 与 19 个结构化断言用例

dlp_io.py

@@ -4,20 +4,24 @@ from __future__ import annotations
 
 import atexit
 import builtins
+import errno
+import hashlib
 import io
 import json
 import operator
 import os
 import queue
+import re
 import shutil
 import subprocess
+import tempfile
 import threading
 from dataclasses import dataclass
 from enum import Enum
 from multiprocessing.context import AuthenticationError
 from multiprocessing.connection import Client
 
-__version__ = "0.1.1"
+__version__ = "0.2.0"
 
 __all__ = [
     "DlpConfig",
@@ -28,9 +32,12 @@ __all__ = [
     "DlpSession",
     "DlpSessionBusyError",
     "DlpTransportError",
+    "DlpWriteError",
+    "WriteChannel",
     "configure",
     "install_open_patch",
     "install_reader_patch",
+    "is_encrypted",
     "open",
     "shutdown",
     "uninstall_open_patch",
@@ -65,6 +72,10 @@ class DlpSessionBusyError(DlpIoError):
     """A session already owns an active file stream."""
 
 
+class DlpWriteError(DlpIoError):
+    """No available write channel can produce an unencrypted file."""
+
+
 # ---------------------------------------------------------------------------
 # Configuration
 
@@ -797,7 +808,7 @@ def open(
         path = os.fsdecode(os.fspath(file))
         raw = _get_default_session().open_raw(path)
         return _wrap_reader(raw, mode, buffering, encoding, errors, newline)
-    return _ORIGINAL_OPEN(file, mode, buffering, encoding, errors, newline, closefd)
+    return _open_write(file, mode, buffering, encoding, errors, newline, closefd)
 
 
 def shutdown() -> None:
@@ -865,7 +876,7 @@ def _reader_wrapper(open_read_func):
                 newline=newline,
                 line_buffering=buffering == 1,
             )
-        return _ORIGINAL_OPEN(file, mode, buffering, encoding, errors, newline, closefd)
+        return _open_write(file, mode, buffering, encoding, errors, newline, closefd)
 
     return patched_open
 
@@ -901,3 +912,282 @@ def uninstall_open_patch() -> None:
         builtins.open = _ORIGINAL_OPEN
         io.open = _ORIGINAL_IO_OPEN
         _installed = False
+
+
+# ---------------------------------------------------------------------------
+# Encryption detection and unencrypted writes
+#
+# DLP transparent encryption stores ciphertext on disk. Approved processes
+# (the whitelisted Python) see plaintext through the filter driver, while
+# non-whitelisted tools such as certutil see the raw on-disk bytes. File
+# attributes (attrib, ADS, size) are identical between encrypted and plain
+# files, so content comparison is the only reliable discriminator.
+
+_DETECT_HEAD_SIZE = 64 * 1024
+_RAW_HASH_PATTERN = re.compile(rb"^[ \t]*([0-9a-fA-F]{64})[ \t\r]*$", re.MULTILINE)
+
+
+def _raw_sha256(path) -> bytes:
+    """SHA-256 of the raw on-disk bytes, read by non-whitelisted certutil.
+
+    certutil must be spawned through ``cmd /c``: DLP hosts block approved
+    processes from starting certutil directly, and list-form arguments keep
+    non-ASCII paths intact through the cmd command line.
+    """
+    try:
+        completed = subprocess.run(
+            ["cmd", "/c", "certutil", "-hashfile", path, "SHA256"],
+            capture_output=True,
+            **_helper_process_options(),
+        )
+    except (OSError, subprocess.SubprocessError) as exc:
+        raise DlpIoError("raw certutil hash unavailable: %s" % exc) from exc
+    if completed.returncode != 0:
+        raise DlpIoError(
+            "certutil -hashfile failed with exit code %d" % completed.returncode
+        )
+    match = _RAW_HASH_PATTERN.search(completed.stdout)
+    if match is None:
+        raise DlpIoError("certutil output did not contain a SHA-256 digest")
+    return bytes.fromhex(match.group(1).decode("ascii"))
+
+
+def _native_sha256(path) -> bytes:
+    digest = hashlib.sha256()
+    with _ORIGINAL_OPEN(path, "rb") as stream:
+        for chunk in iter(lambda: stream.read(CHUNK_SIZE), b""):
+            digest.update(chunk)
+    return digest.digest()
+
+
+def _native_head(path, size) -> bytes:
+    with _ORIGINAL_OPEN(path, "rb") as stream:
+        return stream.read(size)
+
+
+def _helper_head(path, size) -> bytes:
+    raw = _get_default_session().open_raw(path)
+    try:
+        return raw.read(size)
+    finally:
+        raw.close()
+
+
+def is_encrypted(path) -> bool:
+    """Return True when the file on disk holds DLP ciphertext.
+
+    The raw on-disk digest (certutil) is compared with what the current
+    process sees. A mismatch means the file is encrypted and the current
+    process is approved. When both match the result is ambiguous — the file
+    is either plain, or encrypted while the current process is not approved
+    (packaged .exe) — so the first bytes seen by the approved helper are
+    compared with the first bytes seen by the current process.
+    """
+    if os.name != "nt":
+        return False
+    normalized = os.fsdecode(os.fspath(path))
+    native_hash = _native_sha256(normalized)
+    raw_hash = _raw_sha256(normalized)
+    if raw_hash != native_hash:
+        return True
+    return _helper_head(normalized, _DETECT_HEAD_SIZE) != _native_head(
+        normalized, _DETECT_HEAD_SIZE
+    )
+
+
+class WriteChannel(str, Enum):
+    """How path-based writes are routed to keep files unencrypted."""
+
+    DIRECT = "direct"
+    POWERSHELL = "powershell"
+
+
+_WRITE_CHANNEL_ENV = "DLP_IO_WRITE_CHANNEL"
+_WRITE_PROBE_PAYLOAD = b"dlp-io-write-probe\x00\x1a\xff" + bytes(range(256)) * 2
+_write_channel_lock = threading.RLock()
+_write_channel_cache = {}
+
+
+def _powershell_script(path, file_mode) -> str:
+    escaped = path.replace("'", "''")
+    return (
+        "$ErrorActionPreference='Stop';"
+        "$input_stream=[Console]::OpenStandardInput();"
+        "$file_stream=[IO.File]::Open('" + escaped + "',[IO.FileMode]::" + file_mode + ");"
+        "$input_stream.CopyTo($file_stream);"
+        "$file_stream.Close()"
+    )
+
+
+class _PowerShellRelayRaw(io.RawIOBase):
+    """Binary stream written to disk by a non-whitelisted PowerShell child.
+
+    Bytes travel through the stdin pipe (never through a DLP-encrypted temp
+    file), and the child's file writes are not encrypted because the child
+    is outside the DLP write-encryption policy.
+    """
+
+    def __init__(self, path, file_mode) -> None:
+        super().__init__()
+        try:
+            self._proc = subprocess.Popen(
+                ["powershell", "-NoProfile", "-Command", _powershell_script(path, file_mode)],
+                stdin=subprocess.PIPE,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.PIPE,
+                **_helper_process_options(),
+            )
+        except OSError as exc:
+            raise DlpWriteError("failed to start powershell write relay: %s" % exc) from exc
+
+    def writable(self) -> bool:
+        return True
+
+    def write(self, data):
+        if self.closed:
+            raise ValueError("I/O operation on closed file")
+        try:
+            self._proc.stdin.write(data)
+        except (BrokenPipeError, OSError) as exc:
+            raise DlpWriteError("powershell write relay rejected data: %s" % exc) from exc
+        return len(data)
+
+    def flush(self) -> None:
+        if self.closed:
+            raise ValueError("I/O operation on closed file")
+        stdin = self._proc.stdin
+        if stdin is None or stdin.closed:
+            return
+        try:
+            stdin.flush()
+        except (BrokenPipeError, OSError) as exc:
+            raise DlpWriteError("powershell write relay rejected data: %s" % exc) from exc
+
+    def close(self) -> None:
+        if self.closed:
+            return
+        try:
+            try:
+                self._proc.stdin.close()
+            except (BrokenPipeError, OSError):
+                pass
+            returncode = self._proc.wait()
+            if returncode != 0:
+                detail = self._proc.stderr.read().decode("utf-8", "replace").strip()
+                raise DlpWriteError(
+                    "powershell write relay failed with exit code %d: %s"
+                    % (returncode, detail)
+                )
+        finally:
+            super().close()
+
+
+def _write_probe_path(suffix) -> str:
+    descriptor, probe = tempfile.mkstemp(prefix="dlp-io-probe-", suffix=suffix or None)
+    os.close(descriptor)
+    return probe
+
+
+def _probe_direct_write(suffix) -> bool:
+    probe = _write_probe_path(suffix)
+    try:
+        with _ORIGINAL_OPEN(probe, "wb") as stream:
+            stream.write(_WRITE_PROBE_PAYLOAD)
+        try:
+            raw_hash = _raw_sha256(probe)
+        except DlpIoError:
+            # Without a raw channel there is no DLP filter to bypass.
+            return True
+        return raw_hash == hashlib.sha256(_WRITE_PROBE_PAYLOAD).digest()
+    finally:
+        os.unlink(probe)
+
+
+def _probe_powershell_write(suffix) -> bool:
+    probe = _write_probe_path(suffix)
+    try:
+        try:
+            with _PowerShellRelayRaw(probe, "Create") as stream:
+                stream.write(_WRITE_PROBE_PAYLOAD)
+            return _raw_sha256(probe) == hashlib.sha256(_WRITE_PROBE_PAYLOAD).digest()
+        except (DlpIoError, OSError):
+            return False
+    finally:
+        os.unlink(probe)
+
+
+def _select_write_channel(suffix) -> WriteChannel:
+    """Pick a write channel that lands unencrypted bytes on disk.
+
+    Results are cached per file suffix because DLP write policies are
+    typically scoped by document type. ``DLP_IO_WRITE_CHANNEL`` forces a
+    channel and skips probing.
+    """
+    override = os.environ.get(_WRITE_CHANNEL_ENV)
+    if override:
+        try:
+            return WriteChannel(override.strip().lower())
+        except ValueError as exc:
+            valid = ", ".join(channel.value for channel in WriteChannel)
+            raise DlpConfigurationError(
+                "%s must be one of: %s" % (_WRITE_CHANNEL_ENV, valid)
+            ) from exc
+    key = suffix.lower()
+    with _write_channel_lock:
+        cached = _write_channel_cache.get(key)
+        if cached is not None:
+            return cached
+        if _probe_direct_write(suffix):
+            channel = WriteChannel.DIRECT
+        elif _probe_powershell_write(suffix):
+            channel = WriteChannel.POWERSHELL
+        else:
+            raise DlpWriteError(
+                "no write channel can keep %r files unencrypted on this host"
+                % (suffix or "<none>")
+            )
+        _write_channel_cache[key] = channel
+        return channel
+
+
+def _wrap_writer(raw, mode, buffering, encoding, errors, newline):
+    binary = "b" in mode
+    if buffering == 0:
+        if not binary:
+            raw.close()
+            raise ValueError("can't have unbuffered text I/O")
+        return raw
+    buffer_size = _DEFAULT_BUFFER_SIZE if buffering in (-1, 1) else buffering
+    if buffer_size <= 0:
+        raw.close()
+        raise ValueError("invalid buffering size")
+    buffered = io.BufferedWriter(raw, buffer_size=buffer_size)
+    if binary:
+        return buffered
+    return io.TextIOWrapper(
+        buffered,
+        encoding=encoding,
+        errors=errors,
+        newline=newline,
+        line_buffering=buffering == 1,
+    )
+
+
+def _open_write(file, mode, buffering, encoding, errors, newline, closefd):
+    """Open a path for writing so the file stays unencrypted on disk."""
+    path = os.fsdecode(os.fspath(file))
+    if "x" in mode and os.path.exists(path):
+        raise FileExistsError(errno.EEXIST, "File exists", path)
+    channel = _select_write_channel(os.path.splitext(path)[1])
+    if channel is WriteChannel.DIRECT:
+        return _ORIGINAL_OPEN(file, mode, buffering, encoding, errors, newline, closefd)
+    if not closefd:
+        raise ValueError("Cannot use closefd=False with file name")
+    if "w" in mode:
+        file_mode = "Create"
+    elif "a" in mode:
+        file_mode = "Append"
+    else:
+        file_mode = "CreateNew"
+    raw = _PowerShellRelayRaw(path, file_mode)
+    return _wrap_writer(raw, mode, buffering, encoding, errors, newline)