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 个结构化断言用例

.gitea/workflows/publish-dlp-io.yaml

@@ -326,9 +326,9 @@ jobs:
 
           foreach ($artifact in $artifacts) {
             $existing = @($release.assets | Where-Object name -eq $artifact.Name)
-            foreach ($asset in $existing) {
+            if ($existing.Count -gt 0) {
-              Invoke-RestMethod -Method Delete -Headers $headers -Uri "$api/releases/assets/$($asset.id)" | Out-Null
+              Write-Host "asset already exists, skipping: $($artifact.Name)"
-              Write-Host "replaced asset: $($artifact.Name)"
+              continue
             }
             Invoke-RestMethod -Method Post -Headers $headers -Uri "$api/releases/$($release.id)/assets?name=$($artifact.Name)" -Form @{ attachment = Get-Item $artifact.FullName } | Out-Null
             Write-Host "uploaded asset: $($artifact.Name)"
@@ -346,12 +346,12 @@ jobs:
         shell: pwsh
         env:
           TWINE_USERNAME: antior
-          TWINE_PASSWORD: ${{ secrets.GITHUB_TOKEN }}
+          TWINE_PASSWORD: ${{ secrets.CI_PACKAGE_TOKEN }}
           TWINE_REPOSITORY_URL: ${{ gitea.server_url }}/api/packages/antior/pypi
         run: |
           $ErrorActionPreference = "Stop"
           if ([string]::IsNullOrWhiteSpace($env:TWINE_PASSWORD)) {
-            throw "GITHUB_TOKEN is required to publish packages"
+            throw "CI_PACKAGE_TOKEN repository secret (PAT with write:package) is required"
           }
           $headers = @{ Authorization = "token $env:TWINE_PASSWORD" }
           $simpleUrl = "$env:TWINE_REPOSITORY_URL/simple/dlp-io/"

.kimi-code/skills/git-push-dlp/SKILL.md

@@ -0,0 +1,68 @@
+---
+name: git-push-dlp
+description: 在 DLP 透明加密保护区（如 E:\WorkSpace\）内执行 git push。保护区内直接 git push 必现 "fatal: not a git repository (or any of the parent directories): .git"（退出码 128），但其他 git 命令（status / fetch / commit / pull）全部正常。触发条件："git push 失败"、"push 报 not a git repository"、"DLP push"、"git_push_dlp"，或遇到上述 128 错误时。
+---
+
+# git-push-dlp — DLP 保护区内 git push
+
+## 问题背景
+
+本机装有透明加密系统（DLP），`E:\WorkSpace\` 属于保护区。在保护区内执行
+`git push` 会必现：
+
+```
+fatal: not a git repository (or any of the parent directories): .git
+```
+
+退出码 128。这是 DLP 驱动对 push 进程访问模式的拦截，**不是仓库损坏**——
+不要修仓库、不要重装 git。同一个 `.git` 复制到保护区外 push 立刻正常；
+对保护区内的 `.git` 原地解密无效（写回会被驱动重新加密）。
+
+## 原理
+
+1. `robocopy /MIR .git` 到固定 stage 目录 `%TEMP%\git_push_dlp\<repo>\`（保护区外，
+   授权进程读出明文，写出不再加密）。stage 目录跨运行保留：首次全量镜像，
+   之后每次 push 只复制变动的文件。默认通过 `/XD lfs` 排除 `.git\lfs`（LFS
+   对象缓存，动辄数 GB，普通提交 push 不需要）。
+2. 在 stage 副本里执行 `git push`。
+3. 回到原仓库 `git fetch` 同步远程跟踪引用。stage 目录保留不删，供下次增量复用。
+
+## 使用方法
+
+脚本位置：`<skill_dir>\scripts\git_push_dlp.bat`，在**仓库根目录**执行：
+
+```bat
+cmd /c <skill_dir>\scripts\git_push_dlp.bat                  = git push origin HEAD
+cmd /c <skill_dir>\scripts\git_push_dlp.bat origin main      = git push origin main
+cmd /c <skill_dir>\scripts\git_push_dlp.bat upstream main:main
+```
+
+如果本次 push 包含新增 / 变更的 LFS 文件（`.pt` / `.onnx` / `.zip` 等，见
+`.gitattributes`），先设置环境变量再运行，恢复全量复制 `.git\lfs`：
+
+```bat
+set GIT_PUSH_DLP_WITH_LFS=1
+```
+
+## 排障
+
+- **`[ERROR] robocopy .git failed`**：常见于 TEMP 所在盘空间不足。stage 目录
+  跨运行复用（`%TEMP%\git_push_dlp\<repo>\`），旧版本脚本留下的 `<repo>-<数字>`
+  残留目录可以整体删除。仍不足时确认是否误设了
+  `GIT_PUSH_DLP_WITH_LFS`（全量 lfs 可能超过盘符剩余空间）。
+- **stage 在 E: 盘也报 128**：整个 E: 盘都在 DLP 拦截范围内，stage 必须落在
+  C:（`%TEMP%` 默认位置），不要把 `TEMP` 指到 E:。
+- **注释行被当成命令执行 / 解析错乱**：bat 被写成了 LF 行尾或混入了 UTF-8
+  中文注释，见下面维护注意。
+
+## 维护注意（修改 scripts/git_push_dlp.bat 时必读）
+
+- 文件必须保持 **CRLF 行尾**；编辑工具若写成 LF，cmd 会把注释行当命令执行。
+  修正方法：`unix2dos <file>`。
+- 注释（`rem`）保持**英文**。UTF-8 无 BOM 的中文注释会被 cmd 批解析器误解析。
+- 括号代码块内引用本块内 `set` 的变量要用延迟扩展 `!VAR!` 而不是 `%VAR%`。
+
+## 备注
+
+全局副本 `C:\Users\p40000043244\bin\git_push_dlp.bat`（已加入用户 PATH）仍然
+可用，供其他保护区仓库使用；本仓库内以 `<skill_dir>\scripts\` 这份为准。

.kimi-code/skills/git-push-dlp/scripts/git_push_dlp.bat

@@ -0,0 +1,83 @@
+@echo off
+rem ==========================================================================
+rem git_push_dlp.bat
+rem Push from a DLP-protected (transparent-encrypted) directory where
+rem "git push" fails with: fatal: not a git repository
+rem
+rem How it works:
+rem   1. Mirror .git to a fixed per-repo stage dir under %TEMP% with
+rem      robocopy /MIR (outside the protected zone, an authorized process
+rem      reads plaintext, no re-encryption outside). The stage dir is kept
+rem      between runs, so after the first full mirror only changed files
+rem      are copied on subsequent pushes.
+rem   2. Run "git push" from the stage dir.
+rem   3. Run "git fetch" in the original repo to sync remote-tracking refs.
+rem
+rem Usage (run inside the repo root):
+rem   git_push_dlp                      = git push origin HEAD
+rem   git_push_dlp origin main          = git push origin main
+rem   git_push_dlp upstream main:main   = any normal push args
+rem
+rem Env vars:
+rem   GIT_PUSH_DLP_WITH_LFS=1           = also mirror .git\lfs (needed only
+rem                                       when the push contains new LFS
+rem                                       objects)
+rem ==========================================================================
+setlocal EnableExtensions EnableDelayedExpansion
+chcp 65001 >nul
+
+if not exist ".git\" (
+    echo [ERROR] .git not found. Run this inside a git repository root.
+    exit /b 1
+)
+
+for %%I in ("%CD%") do set "REPO_NAME=%%~nxI"
+set "STAGE=%TEMP%\git_push_dlp\%REPO_NAME%"
+
+echo [DLP] Stage dir: %STAGE%
+
+rem .git\lfs is the LFS object cache (often several GB); normal commits do
+rem not need it for push. Excluding it avoids filling up the TEMP drive.
+rem If this push contains new/changed LFS files, set GIT_PUSH_DLP_WITH_LFS=1 first.
+set "EXCLUDE_OPT=/XD lfs"
+if defined GIT_PUSH_DLP_WITH_LFS set "EXCLUDE_OPT="
+if defined GIT_PUSH_DLP_WITH_LFS (
+    echo [DLP] Including .git\lfs
+) else (
+    echo [DLP] Excluding .git\lfs ^(set GIT_PUSH_DLP_WITH_LFS=1 to include^)
+)
+
+robocopy ".git" "%STAGE%\.git" /MIR %EXCLUDE_OPT% /NFL /NDL /NJH /NP >nul
+if errorlevel 8 (
+    echo [ERROR] robocopy .git failed ^(exit %ERRORLEVEL%^).
+    exit /b 1
+)
+
+set "REMOTE=%~1"
+if not defined REMOTE set "REMOTE=origin"
+if "%REMOTE:~0,1%"=="-" set "REMOTE=origin"
+
+set "PUSHARGS=%*"
+if not defined PUSHARGS set "PUSHARGS=origin HEAD"
+
+pushd "%STAGE%"
+echo [DLP] Run: git push %PUSHARGS%
+git push %PUSHARGS%
+set "EXIT_CODE=%ERRORLEVEL%"
+popd
+
+if not "%EXIT_CODE%"=="0" (
+    echo [ERROR] push failed with exit code %EXIT_CODE%
+    exit /b %EXIT_CODE%
+)
+
+echo [DLP] Sync back: git fetch %REMOTE%
+git fetch %REMOTE%
+set "FETCH_CODE=%ERRORLEVEL%"
+
+if not "%FETCH_CODE%"=="0" (
+    echo [WARN] push succeeded but fetch back failed with exit code %FETCH_CODE%
+    exit /b %FETCH_CODE%
+)
+echo [DLP] Done.
+exit /b 0

AGENTS.md

@@ -30,3 +30,15 @@
 - 项目内所有 `test_*.py`、`*_test.py` 和 `conftest.py` 都必须通过文本断言 AST policy gate，包括根测试集之外的子项目测试。
 - 新测试和既有测试适用同一规则，不允许 grandfathered violation。
 - 字符串可以作为被测 API 输入、Path 构造参数、字典 key 或 mock 调用参数；禁止的是对纯文本结果、渲染文本、源码文本、日志文本和异常消息进行断言。
+
+## DLP 与 git push
+
+本仓库位于 DLP 透明加密保护区（`E:\WorkSpace\`）。在保护区内直接执行 `git push` 会必现 `fatal: not a git repository (or any of the parent directories): .git`（退出码 128）；其他 git 命令（status / commit / pull 等）全部正常，这是 DLP 驱动对 push 进程访问模式的拦截，**不是仓库损坏**，不要修仓库、不要重装 git。
+
+提交推送（git push）必须调用项目内 skill `git-push-dlp`（`.kimi-code/skills/git-push-dlp/`，详细用法见其 `SKILL.md`）：
+
+```bat
+cmd /c .kimi-code\skills\git-push-dlp\scripts\git_push_dlp.bat origin main
+```
+
+push 包含新增 / 变更的 LFS 对象时，先 `set GIT_PUSH_DLP_WITH_LFS=1` 再运行（恢复全量复制 `.git\lfs`）。

README.md

@@ -4,6 +4,22 @@ io-compatible file reads through an approved Python helper on Windows DLP hosts.
 
 完整文档见 [docs/DLP_IO_LIBRARY.md](docs/DLP_IO_LIBRARY.md)。
 
+## 安装
+
+从 Gitea PyPI registry 安装（匿名可下载，无需 token）：
+
+```powershell
+py -m pip install --index-url https://gitea.docker.antior.cn/api/packages/antior/pypi/simple dlp-io
+```
+
+pip 会自动选择最匹配的 wheel：Windows 上 CPython 3.10–3.14 安装对应的 pyd wheel（整个库由 Cython 编译为单个 `.pyd`），其余环境安装纯 Python wheel。如需固定版本：
+
+```powershell
+py -m pip install --index-url https://gitea.docker.antior.cn/api/packages/antior/pypi/simple dlp-io==0.1.1
+```
+
+也可以从 [Releases](https://gitea.docker.antior.cn/antior/dlp-io/releases) 页面下载产物：wheel 用 `py -m pip install <文件>` 安装；`dlp_io.py` 或对应版本的裸 `.pyd` 为免安装单文件，直接放进项目目录（或加入 `PYTHONPATH`）即可 `import dlp_io`。
+
 ## 开发
 
 ```bash

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)

docs/DLP_IO_LIBRARY.md

@@ -1,8 +1,8 @@
 # dlp-io
 
-`dlp-io` 为 Windows DLP 透明加密环境提供接近 `io.open` 的 Python 文件 API。非白名单进程负责业务逻辑和写入；文件读取由已获 DLP 白名单授权的 Python helper 完成，再通过经过认证的 Windows Named Pipe 返回明文字节。
+`dlp-io` 为 Windows DLP 透明加密环境提供接近 `io.open` 的 Python 文件 API。非白名单进程负责业务逻辑；文件读取由已获 DLP 白名单授权的 Python helper 完成，再通过经过认证的 Windows Named Pipe 返回明文字节；文件写入通过通道探测保证落盘为未加密格式。另提供 `is_encrypted()`，在读取前判断文件是否处于 DLP 加密状态。
 
-当前版本为 `dlp-io==0.1.1`，distribution 名是 `dlp-io`，import 名是 `dlp_io`。
+当前版本为 `dlp-io==0.2.0`，distribution 名是 `dlp-io`，import 名是 `dlp_io`。
 
 ## 安装
 
@@ -11,7 +11,7 @@
 registry 已开放匿名下载，一条命令即可安装；pip 会自动选择最匹配的 wheel：Windows 上 CPython 3.10–3.14 会得到对应的 pyd wheel（整个库编译为单个 `.pyd`），其余环境得到纯 Python wheel。
 
 ```powershell
-py -m pip install --index-url https://gitea.docker.antior.cn/api/packages/antior/pypi/simple dlp-io==0.1.1
+py -m pip install --index-url https://gitea.docker.antior.cn/api/packages/antior/pypi/simple dlp-io==0.2.0
 ```
 
 Package 页面：<https://gitea.docker.antior.cn/antior/-/packages/pypi/dlp-io>
@@ -78,14 +78,50 @@ dlp_io.shutdown()
 
 模式路由规则：
 
-- 路径型 `r`、`rt`、`rb`：通过 helper 读取。
+- 路径型 `r`、`rt`、`rb`：通过 helper 读取，始终返回明文。
-- 路径型 `w`、`a`、`x`：使用原生 `open` 写入。
+- 路径型 `w`、`a`、`x`：经过写入通道路由（见下文「未加密写入」），保证落盘为未加密文件。
 - 整数文件描述符或自定义 `opener`：使用原生 `open`。
 - 路径型 `+` 更新模式：显式抛出 `io.UnsupportedOperation`。
 - 路径读取配合 `closefd=False`：与标准 API 一样抛出 `ValueError`。
 - `rb` 配合 `buffering=0`：返回 raw stream；默认返回 buffered reader。
 - 文本模式的默认 encoding 与当前 Python `io.open` 一致；跨机器文件请显式传 `encoding="utf-8"`。
 
+## 加密状态判断
+
+`is_encrypted(path)` 在读取前判断文件是否处于 DLP 加密状态，返回 `bool`：
+
+```python
+import dlp_io
+
+if dlp_io.is_encrypted(r"D:\Protected\input.pptx"):
+    # 已加密：当前进程若是打包的 EXE（非白名单），必须经 helper 中转读取
+    with dlp_io.open(r"D:\Protected\input.pptx", "rb") as stream:
+        data = stream.read()
+else:
+    # 未加密：可用任意方式直接读取
+    with dlp_io.open(r"D:\Protected\input.pptx", "rb") as stream:
+        data = stream.read()
+```
+
+判断原理：DLP 透明加密在磁盘上存密文，白名单进程读到明文、非白名单工具（certutil）读到原始盘内字节；加密文件与未加密文件的 attrib 属性、ADS、大小完全一致，只有内容可区分。`is_encrypted()` 先比较 certutil 原始哈希与当前进程视图哈希：不一致则文件已加密且当前进程是白名单；一致时存在歧义（文件未加密，或文件已加密但当前进程是非白名单 EXE），再比较 helper 视图与当前进程视图的前 64 KiB 定案。两种运行环境（Python/pyd 与打包 EXE）结果都正确。
+
+代价：未加密文件需要一次完整 certutil 哈希、一次本地完整读取和一次 helper 前 64 KiB 读取；大文件请缓存判断结果，不要每次读取前重复调用。helper 未启动时会复用懒加载的默认 session，结束后照常 `dlp_io.shutdown()`。非 Windows 平台直接返回 `False`。
+
+## 未加密写入
+
+写入目标始终是「磁盘上保存未加密文件」：
+
+- 打包为 EXE（非白名单进程）：原生直写天然落盘明文，选中直写通道。
+- Python 调试 / pyd（白名单进程）：部分 DLP 策略会加密白名单进程的写入。首次写某个后缀时库会用临时探针文件实测直写是否落盘明文（用 certutil 原始哈希校验）；若直写会被加密，自动改走 PowerShell 中转通道——字节经 stdin 管道传给非白名单的 powershell.exe 子进程写盘，不经过任何磁盘临时文件。通道按文件后缀缓存；两个通道都不可用时抛出 `DlpWriteError`，绝不静默写出密文。
+
+可用环境变量跳过探测、强制指定通道（值为 `direct` 或 `powershell`）：
+
+```powershell
+$env:DLP_IO_WRITE_CHANNEL = "powershell"
+```
+
+探测按后缀缓存，注意 DLP 策略若按目录区分，探针结果可能与目标目录不同；此时建议用环境变量显式指定通道。
+
 ## 配置和生命周期
 
 默认 session 在第一次读取时懒启动，并被后续读取复用：
@@ -154,6 +190,7 @@ finally:
 - `DlpProtocolError`：协议版本、控制消息或响应结构无效。
 - `DlpTransportError`：helper 在显式 EOF 前断开，或文件传输被截断。
 - `DlpSessionBusyError`：同一个 session 已有活动 stream。
+- `DlpWriteError`：没有任何写入通道能落盘未加密文件，或 PowerShell 中转写入失败。
 
 helper 启动或读取失败时，库会显式报错并且不回退到当前 EXE 直接读取。DLP 直读可能返回合法长度的密文；静默回退会把数据损坏伪装成成功。
 

tests/test_dlp_io_encryption.py

@@ -0,0 +1,262 @@
+from __future__ import annotations
+
+import hashlib
+import os
+from pathlib import Path
+
+import pytest
+
+import dlp_io
+
+
+pytestmark = pytest.mark.skipif(os.name != "nt", reason="DLP detection requires Windows")
+
+SAMPLES_DIR = Path(__file__).resolve().parent
+ENCRYPTED_SAMPLE = SAMPLES_DIR / "DLP加密文件.pptx"
+DECRYPTED_SAMPLE = SAMPLES_DIR / "DLP解密文件.pptx"
+
+
+@pytest.fixture(autouse=True)
+def reset_module_state(monkeypatch):
+    monkeypatch.delenv(dlp_io._WRITE_CHANNEL_ENV, raising=False)
+    monkeypatch.setattr(dlp_io, "_config", dlp_io.DlpConfig())
+    dlp_io._write_channel_cache.clear()
+    dlp_io.shutdown()
+    yield
+    dlp_io._write_channel_cache.clear()
+    dlp_io.shutdown()
+
+
+def _native_sha256(path: Path) -> bytes:
+    return hashlib.sha256(path.read_bytes()).digest()
+
+
+def _dlp_host() -> bool:
+    """本机是否启用 DLP：加密样本的原始盘内字节与本地视图是否不一致。"""
+    return dlp_io._raw_sha256(str(ENCRYPTED_SAMPLE)) != _native_sha256(ENCRYPTED_SAMPLE)
+
+
+# ---------------------------------------------------------------------------
+# is_encrypted against the real DLP samples
+
+
+def test_encrypted_sample_matches_host_dlp_capability() -> None:
+    assert dlp_io.is_encrypted(ENCRYPTED_SAMPLE) == _dlp_host()
+
+
+def test_decrypted_sample_is_never_reported_encrypted() -> None:
+    assert dlp_io.is_encrypted(DECRYPTED_SAMPLE) is False
+
+
+def test_is_encrypted_missing_file_raises() -> None:
+    with pytest.raises(FileNotFoundError):
+        dlp_io.is_encrypted(SAMPLES_DIR / "missing-sample.bin")
+
+
+# ---------------------------------------------------------------------------
+# is_encrypted decision branches (mocked channels)
+
+
+def test_is_encrypted_short_circuits_when_raw_digest_differs(monkeypatch) -> None:
+    monkeypatch.setattr(dlp_io, "_native_sha256", lambda path: b"\x01" * 32)
+    monkeypatch.setattr(dlp_io, "_raw_sha256", lambda path: b"\x02" * 32)
+    monkeypatch.setattr(
+        dlp_io,
+        "_get_default_session",
+        lambda: pytest.fail("helper must not start when digests differ"),
+    )
+
+    assert dlp_io.is_encrypted("payload.bin") is True
+
+
+@pytest.mark.parametrize(
+    ("helper_head", "native_head", "expected"),
+    [
+        (b"\x50\x4b\x03\x04", b"\x18\x1b\x03\x1a", True),
+        (b"\x50\x4b\x03\x04", b"\x50\x4b\x03\x04", False),
+    ],
+)
+def test_is_encrypted_ambiguous_case_compares_helper_head(
+    monkeypatch, helper_head: bytes, native_head: bytes, expected: bool
+) -> None:
+    digest = b"\x01" * 32
+    monkeypatch.setattr(dlp_io, "_native_sha256", lambda path: digest)
+    monkeypatch.setattr(dlp_io, "_raw_sha256", lambda path: digest)
+    monkeypatch.setattr(dlp_io, "_native_head", lambda path, size: native_head)
+    monkeypatch.setattr(dlp_io, "_helper_head", lambda path, size: helper_head)
+
+    assert dlp_io.is_encrypted("payload.bin") is expected
+
+
+# ---------------------------------------------------------------------------
+# Write channel selection
+
+
+def test_select_write_channel_honours_probes(monkeypatch, tmp_path) -> None:
+    monkeypatch.setattr(dlp_io, "_probe_direct_write", lambda suffix: True)
+    monkeypatch.setattr(
+        dlp_io,
+        "_probe_powershell_write",
+        lambda suffix: pytest.fail("powershell probe must be skipped"),
+    )
+    assert dlp_io._select_write_channel(".bin") is dlp_io.WriteChannel.DIRECT
+
+    monkeypatch.setattr(dlp_io, "_probe_direct_write", lambda suffix: False)
+    monkeypatch.setattr(dlp_io, "_probe_powershell_write", lambda suffix: True)
+    assert dlp_io._select_write_channel(".pptx") is dlp_io.WriteChannel.POWERSHELL
+
+
+def test_select_write_channel_raises_when_no_channel_works(monkeypatch) -> None:
+    monkeypatch.setattr(dlp_io, "_probe_direct_write", lambda suffix: False)
+    monkeypatch.setattr(dlp_io, "_probe_powershell_write", lambda suffix: False)
+
+    with pytest.raises(dlp_io.DlpWriteError):
+        dlp_io._select_write_channel(".bin")
+
+
+def test_select_write_channel_caches_per_suffix(monkeypatch) -> None:
+    calls = []
+    monkeypatch.setattr(
+        dlp_io, "_probe_direct_write", lambda suffix: calls.append(suffix) or True
+    )
+
+    assert dlp_io._select_write_channel(".BIN") is dlp_io.WriteChannel.DIRECT
+    assert dlp_io._select_write_channel(".bin") is dlp_io.WriteChannel.DIRECT
+    assert len(calls) == 1
+
+
+def test_write_channel_env_override(monkeypatch) -> None:
+    monkeypatch.setenv(dlp_io._WRITE_CHANNEL_ENV, "PowerShell")
+    monkeypatch.setattr(
+        dlp_io,
+        "_probe_direct_write",
+        lambda suffix: pytest.fail("override must skip probing"),
+    )
+    assert dlp_io._select_write_channel(".bin") is dlp_io.WriteChannel.POWERSHELL
+
+    monkeypatch.setenv(dlp_io._WRITE_CHANNEL_ENV, "bogus")
+    with pytest.raises(dlp_io.DlpConfigurationError):
+        dlp_io._select_write_channel(".bin")
+
+
+def test_real_probe_selects_a_working_channel(tmp_path) -> None:
+    try:
+        channel = dlp_io._select_write_channel(".bin")
+    except dlp_io.DlpWriteError:
+        pytest.skip("host cannot produce unencrypted writes")
+
+    direct_ok = dlp_io._probe_direct_write(".bin")
+    expected = dlp_io.WriteChannel.DIRECT if direct_ok else dlp_io.WriteChannel.POWERSHELL
+    assert channel is expected
+
+
+# ---------------------------------------------------------------------------
+# PowerShell relay writer
+
+
+def test_powershell_relay_writes_exact_binary_and_lands_plaintext(tmp_path) -> None:
+    payload = bytes(range(256)) * 64 + b"\x00\x1a\xff"
+    target = tmp_path / "relay输出.bin"
+
+    with dlp_io._PowerShellRelayRaw(str(target), "Create") as raw:
+        assert raw.writable()
+        assert raw.write(payload) == len(payload)
+
+    assert target.read_bytes() == payload
+    assert dlp_io._raw_sha256(str(target)) == hashlib.sha256(payload).digest()
+
+
+def test_powershell_relay_append_preserves_existing_content(tmp_path) -> None:
+    target = tmp_path / "relay.bin"
+
+    with dlp_io._PowerShellRelayRaw(str(target), "Create") as raw:
+        raw.write(b"ab")
+    with dlp_io._PowerShellRelayRaw(str(target), "Append") as raw:
+        raw.write(b"cd")
+
+    assert target.read_bytes() == b"abcd"
+
+
+def test_powershell_relay_create_new_fails_on_existing_file(tmp_path) -> None:
+    target = tmp_path / "relay.bin"
+    target.write_bytes(b"existing")
+
+    with pytest.raises(dlp_io.DlpWriteError):
+        with dlp_io._PowerShellRelayRaw(str(target), "CreateNew"):
+            pass
+
+
+def test_open_write_routes_through_relay_when_selected(monkeypatch, tmp_path) -> None:
+    monkeypatch.setattr(
+        dlp_io, "_select_write_channel", lambda suffix: dlp_io.WriteChannel.POWERSHELL
+    )
+    payload = bytes(range(256)) * 32
+    target = tmp_path / "routed.bin"
+
+    with dlp_io.open(target, "wb") as file:
+        file.write(payload)
+
+    assert target.read_bytes() == payload
+
+
+def test_open_text_write_routes_through_relay(monkeypatch, tmp_path) -> None:
+    monkeypatch.setattr(
+        dlp_io, "_select_write_channel", lambda suffix: dlp_io.WriteChannel.POWERSHELL
+    )
+    payload = "中文内容\n第二行".encode("utf-8")
+    target = tmp_path / "routed.txt"
+
+    with dlp_io.open(target, "w", encoding="utf-8", newline="") as file:
+        file.write(payload.decode("utf-8"))
+    with dlp_io.open(target, "a", encoding="utf-8", newline="") as file:
+        file.write("追加")
+
+    assert target.read_bytes() == payload + "追加".encode("utf-8")
+
+
+def test_open_exclusive_create_is_atomic_before_channel_selection(
+    monkeypatch, tmp_path
+) -> None:
+    monkeypatch.setattr(
+        dlp_io,
+        "_select_write_channel",
+        lambda suffix: pytest.fail("existing target must fail before probing"),
+    )
+    target = tmp_path / "existing.bin"
+    target.write_bytes(b"data")
+
+    with pytest.raises(FileExistsError):
+        dlp_io.open(target, "xb")
+
+    assert target.read_bytes() == b"data"
+
+
+def test_open_write_direct_channel_uses_native_open(monkeypatch, tmp_path) -> None:
+    monkeypatch.setattr(
+        dlp_io, "_select_write_channel", lambda suffix: dlp_io.WriteChannel.DIRECT
+    )
+    target = tmp_path / "direct.bin"
+
+    with dlp_io.open(target, "wb") as file:
+        assert not isinstance(file, dlp_io._PowerShellRelayRaw)
+        file.write(b"native")
+
+    assert target.read_bytes() == b"native"
+
+
+def test_patched_open_write_uses_same_channel_routing(monkeypatch, tmp_path) -> None:
+    import builtins
+
+    monkeypatch.setattr(
+        dlp_io, "_select_write_channel", lambda suffix: dlp_io.WriteChannel.POWERSHELL
+    )
+    target = tmp_path / "patched.bin"
+
+    dlp_io.install_open_patch()
+    try:
+        with builtins.open(target, "wb") as file:
+            file.write(b"patched-write")
+    finally:
+        dlp_io.uninstall_open_patch()
+
+    assert target.read_bytes() == b"patched-write"