feat: 支持 Python 3.8+ (0.3.0)

- requires-python >=3.8, classifiers 增加 3.8/3.9 - 代码经 ast feature_version=(3,8) 校验, 无 3.9+ 专属语法; str|None 等注解由 future annotations 延迟求值 - CI 测试/构建/smoke 矩阵扩展为 3.8-3.14, 裸资产数 6→8 - 注意: Windows runner 需预装 Python 3.8 与 3.9 (py launcher)
docs: DLP 环境特征调研记录与判定/写入逻辑说明
2026-08-03 15:30:33 +08:00 · 2026-07-31 18:32:14 +08:00 · 2026-07-31 18:24:22 +08:00 · 2026-07-31 18:13:44 +08:00 · 2026-07-31 17:59:19 +08:00 · 2026-07-31 17:29:40 +08:00
13 changed files with 975 additions and 30 deletions
@@ -95,11 +95,11 @@ jobs:
          "DLP_IO_PACKAGE_VERSION=$packageVersion" | Out-File $env:GITHUB_ENV -Encoding utf8 -Append
          Write-Host "package source: $tag @ $headCommit"

-      - name: Test Python 3.10-3.14
+      - name: Test Python 3.8-3.14
        shell: pwsh
        run: |
          $ErrorActionPreference = "Stop"
-          foreach ($pythonVersion in @("3.10", "3.11", "3.12", "3.13", "3.14")) {
+          foreach ($pythonVersion in @("3.8", "3.9", "3.10", "3.11", "3.12", "3.13", "3.14")) {
            & py "-$pythonVersion" --version
            if ($LASTEXITCODE -ne 0) { throw "Python $pythonVersion is required" }
            & py "-$pythonVersion" -m pip install -r requirements-dev.txt --progress-bar off
@@ -124,7 +124,7 @@ jobs:
            "dlp_io-$version-py3-none-any.whl",
            "dlp_io-$version.tar.gz"
          )
-          foreach ($pythonVersion in @("3.10", "3.11", "3.12", "3.13", "3.14")) {
+          foreach ($pythonVersion in @("3.8", "3.9", "3.10", "3.11", "3.12", "3.13", "3.14")) {
            $abi = $pythonVersion.Replace(".", "")
            $wheel = "dlp_io-$version-cp$abi-cp$abi-win_amd64.whl"
            $env:DLP_IO_PYD = "1"
@@ -164,7 +164,7 @@ jobs:
            }
          }
          $bareAssets = @(Get-ChildItem -LiteralPath $assetsDir -File | ForEach-Object Name)
-          if ($bareAssets.Count -ne 6) {
+          if ($bareAssets.Count -ne 8) {
            throw "bare asset set mismatch: $($bareAssets -join ', ')"
          }

@@ -198,7 +198,7 @@ jobs:
            }
          }

-          foreach ($pythonVersion in @("3.10", "3.11", "3.12", "3.13", "3.14")) {
+          foreach ($pythonVersion in @("3.8", "3.9", "3.10", "3.11", "3.12", "3.13", "3.14")) {
            $abi = $pythonVersion.Replace(".", "")
            $pydWheel = (Resolve-Path (Join-Path dist "dlp_io-$($env:DLP_IO_PACKAGE_VERSION)-cp$abi-cp$abi-win_amd64.whl")).Path
            $pydRoot = Join-Path $env:TEMP ("dlp-io-pyd-smoke-$abi-" + $PID)
@@ -232,7 +232,7 @@ jobs:
            throw "unsafe bare smoke path: $bareRootFull"
          }
          try {
-            foreach ($pythonVersion in @("3.10", "3.11", "3.12", "3.13", "3.14")) {
+            foreach ($pythonVersion in @("3.8", "3.9", "3.10", "3.11", "3.12", "3.13", "3.14")) {
              $abi = $pythonVersion.Replace(".", "")
              $sandbox = Join-Path $bareRootFull "cp$abi"
              New-Item -ItemType Directory -Force $sandbox | Out-Null
@@ -293,7 +293,7 @@ jobs:
              "## 产物",
              "",
              "- ``dlp_io-$version-py3-none-any.whl``：纯 Python wheel，pip 安装，跨平台",
-              "- ``dlp_io-$version-cp3XX-cp3XX-win_amd64.whl``：pyd wheel，pip 安装，Windows 按解释器版本选用（cp310-cp314）",
+              "- ``dlp_io-$version-cp3XX-cp3XX-win_amd64.whl``：pyd wheel，pip 安装，Windows 按解释器版本选用（cp38-cp314）",
              "- ``dlp_io-$version.tar.gz``：sdist 源码包",
              "- ``dlp_io.py`` / ``dlp_io.cp3XX-win_amd64.pyd``：免安装单文件，直接放进项目目录即可 ``import dlp_io``",
              "",
@@ -326,9 +326,9 @@ jobs:

          foreach ($artifact in $artifacts) {
            $existing = @($release.assets | Where-Object name -eq $artifact.Name)
-            foreach ($asset in $existing) {
-              Invoke-RestMethod -Method Delete -Headers $headers -Uri "$api/releases/assets/$($asset.id)" | Out-Null
-              Write-Host "replaced asset: $($artifact.Name)"
+            if ($existing.Count -gt 0) {
+              Write-Host "asset already exists, skipping: $($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/"
@@ -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\` 这份为准。
@@ -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
@@ -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`）。
@@ -2,7 +2,46 @@

 io-compatible file reads through an approved Python helper on Windows DLP hosts.

-完整文档见 [docs/DLP_IO_LIBRARY.md](docs/DLP_IO_LIBRARY.md)。
+完整文档见 [docs/DLP_IO_LIBRARY.md](docs/DLP_IO_LIBRARY.md)；DLP 环境实测特征与判定/写入逻辑推导见 [docs/DLP_ENVIRONMENT_NOTES.md](docs/DLP_ENVIRONMENT_NOTES.md)。
+
+## DLP 程序化特征与实现逻辑
+
+在透明加密 DLP 环境下实测到的关键事实（详见上面的特征文档）：
+
+- **文件属性不可区分**：加密/未加密文件的 attrib、ADS、大小完全一致，只有内容可判。
+- **同尺寸流式加密**：密文无附加头尾、无静态 magic，且进程按身份分白名单——
+  `python.exe`/`cmd.exe`/`powershell.exe` 读加密文件得到明文（透明解密），
+  `certutil.exe` 等非白名单工具得到磁盘密文。
+- **白名单读透明 ≠ 写加密**：写入是否被加密因机器策略而异，必须运行时探测。
+
+对应的实现逻辑：
+
+- **读取始终明文**：打包 EXE（非白名单）经 stdin 注入 helper 源码拉起白名单
+  `python.exe`，由它读明文、经认证 Named Pipe 回传（python 按身份白名单，
+  任何父进程拉起都有效）。
+- **`is_encrypted(path)` 加密判定**：先比 certutil 原始哈希与当前进程视图哈希，
+  不等即加密；相等时存在歧义（未加密，或 EXE 读加密文件），再比 helper 视图与
+  本地视图的前 64 KiB 定案——Python/pyd/EXE 三种环境都正确。
+- **写入始终未加密**：按文件后缀探测直写是否落盘明文（写探针文件后用 certutil
+  原始哈希校验）；若直写会被加密，自动切换到 PowerShell stdin 管道中转写盘，
+  字节不落地任何临时文件；两通道均不可用则抛 `DlpWriteError`，绝不静默写密文。
+  可用 `DLP_IO_WRITE_CHANNEL=direct|powershell` 强制指定通道。
+
+## 安装
+
+从 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.8–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`。

 ## 开发

@@ -13,10 +52,10 @@ pytest

 ## 构建发布

-发布流程由 `.gitea/workflows/publish-dlp-io.yaml` 驱动：推送 `dlp-io-vX.Y.Z` 格式的 annotated tag，Windows runner（Python 3.10–3.14）会自动完成测试、构建，并把产物上传到 Gitea Release：
+发布流程由 `.gitea/workflows/publish-dlp-io.yaml` 驱动：推送 `dlp-io-vX.Y.Z` 格式的 annotated tag，Windows runner（Python 3.8–3.14）会自动完成测试、构建，并把产物上传到 Gitea Release：

 - `dlp_io-X.Y.Z-py3-none-any.whl`（纯 Python，跨平台）
- `dlp_io-X.Y.Z-cp310` … `cp314-win_amd64.whl`（5 个 pyd wheel，Cython 编译，Windows 按解释器版本选用）
+- `dlp_io-X.Y.Z-cp38` … `cp314-win_amd64.whl`（7 个 pyd wheel，Cython 编译，Windows 按解释器版本选用）
 - `dlp_io-X.Y.Z.tar.gz`（sdist）
 - `dlp_io.py` 和 5 个裸 `dlp_io.cp3XX-win_amd64.pyd`（免安装单文件，放进项目目录即可 import）

@@ -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.3.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)
@@ -0,0 +1,150 @@
+# DLP 环境特征调研与检测/写入实现逻辑
+
+本文记录在本机（启用透明加密 DLP 的 Windows）上对 `tests/DLP加密文件.pptx` 与
+`tests/DLP解密文件.pptx` 两个样本的实测结论、探索过程，以及 `dlp_io` 据此实现的
+`is_encrypted()` 判定逻辑和未加密写入通道选择逻辑。所有结论都来自实测，不同 DLP
+产品/策略可能不同，库的应对方式是运行时探测而非硬编码假设。
+
+## 一、样本事实
+
+两个样本文件内容相同（解密后 SHA-256 均为
+`968b5db1…d2947100`），大小完全相同（2,466,572 字节）。加密文件在磁盘上存密文，
+前 16 字节对比：
+
+```
+明文视图: 504b0304 0a000000 0000874e e2400000   (标准 ZIP/PPTX 头 "PK\x03\x04")
+密文视图: 181b031a 1510197c 0a19 0e7c 6f726c5f   (流式加密, 相同明文 0x00 在不同偏移密文不同)
+```
+
+特征：同尺寸、全文流式加密、无附加文件头/尾、无静态 magic 可识别。
+
+## 二、文件属性层面无法区分（重要结论）
+
+对两个样本逐项对比，全部一致：
+
+- `attrib`：都只有 `A` 属性（无 EFS 的 E 属性等差异）
+- `dir /r`：均无附加数据流（ADS）
+- 文件大小、时间戳：一致
+
+**结论：无法通过任何文件元数据判断加密状态，唯一可靠的判别特征是内容。**
+这正是 `is_encrypted()` 采用内容对比的原因。
+
+## 三、进程白名单模型（实测矩阵）
+
+DLP 透明加密按「进程身份」决定是否介入读写。用「读取加密样本看到什么」实测：
+
+| 进程 | 父进程 | 读加密样本视图 | 结论 |
+|------|--------|----------------|------|
+| `python.exe` | 任意（含非白名单 bash） | 明文 (PK 头) | **身份白名单**，不继承、不降级 |
+| `cmd.exe`（`type`） | 任意 | 明文 | 白名单 |
+| `powershell.exe` | 任意 | 明文 | 白名单（读透明） |
+| `certutil.exe` | bash / cmd | 密文 | **非白名单， raw 视图** |
+| Git Bash msys 工具（`head`/`sha256sum`/`cat`） | bash | 密文 | 非白名单 |
+| `git.exe`（`hash-object`） | bash | 明文 | 白名单（两样本 blob hash 相同） |
+
+关键推论：
+
+1. **helper 桥接设计成立**：打包 EXE（非白名单）拉起 `python.exe` 子进程，python 按身份
+   仍是白名单，读到的就是明文——这是库读取链路的根基。
+2. **`cmd /c type` 不能当密文通道**：cmd 是白名单，读到的同样是明文。社区直觉「用 cmd
+   读原始内容」在本 DLP 上不成立。
+3. **certutil 是本机可用的 raw 视图工具**：但 DLP 拦截白名单进程直接拉起它
+   （python → certutil 报 `WinError 786`），必须经 `cmd /c certutil …` 中转。
+4. **`cmd /c` 中文路径坑**：把整条命令作为单个字符串传参会触发 cmd 的引号剥离规则，
+   中文文件名丢失（`ERROR_FILE_NOT_FOUND`）；用列表参数（subprocess 自动逐个加引号）
+   则中文路径正常。
+
+写入侧实测：
+
+| 写入方 | 落盘结果（raw 视图校验） |
+|--------|--------------------------|
+| `python.exe` 直写（.bin / .pptx 后缀） | 明文（本机策略不加密 python 写入） |
+| `powershell.exe`（stdin 管道 → `[IO.File]`） | 明文 |
+| `cmd /c more > f` | **破坏二进制**（0x00→CRLF、TAB→空格），不可用 |
+| `cmd /c copy /b con f` | 挂起等待控制台输入，不可用 |
+| `certutil -decode - f` | 不支持 stdin 输入，不可用 |
+| `findstr ^ f` | raw 密文视图，但按行重新包装（0x0A→CRLF 注入），只能看头部不能还原内容 |
+
+注意：白名单读透明 ≠ 写加密。本机 python/powershell 读透明但写入落盘明文；
+「白名单进程写入会被加密」的策略在其他机器上仍可能存在——所以写入通道必须
+运行时探测，不能假设。
+
+## 四、探索过程（方法记录）
+
+1. **多通道哈希对比**：`certutil -hashfile`（非白名单视图）对两样本哈希不同
+   （33d3… vs 968b…），python 读取两样本哈希相同（968b…）——证明 DLP 生效、
+   python 白名单、certutil 非白名单，一条命令同时确认三件事。
+2. **头部字节对比**：`head -c 64 | xxd`（msys 工具，raw 视图）看到密文头，确认
+   同尺寸流加密、无附加头。
+3. **文件属性对比**：`attrib` / `dir /r` 无差异——排除元数据检测路线。
+4. **写入通道候选实验**：逐一验证 more/copy con/certutil -decode/findstr 全部
+   有二进制安全性或可用性问题，最终 `powershell [Console]::OpenStandardInput()`
+   管道方案通过全字节（含 0x00/0x1A/0xFF）哈希校验。
+5. **误报纠正**：`python → cmd /c type` 一度显示明文，先怀疑「白名单继承」，
+   后用 `bash → cmd /c type`（仍是明文）和 `bash → python`（明文）交叉验证，
+   确认 python/cmd 均为身份白名单而非继承。
+
+## 五、is_encrypted() 判定逻辑
+
+判据：文件已加密 ⟺ 白名单视图 ≠ 非白名单（raw）视图。
+
+难点：当前进程自身是否白名单是未知的（Python/pyd 是白名单，打包 EXE 不是），
+单次对比存在歧义。`dlp_io.is_encrypted(path)` 用两阶段消歧：
+
+```
+h_raw    = certutil -hashfile(path)     # 非白名单 raw 视图（经 cmd /c 中转）
+h_native = hashlib(当前进程读 path)
+
+if h_raw != h_native:
+    return True          # 视图不一致: 文件已加密, 且当前进程是白名单
+
+# 歧义分支: 文件未加密, 或文件已加密但当前进程是非白名单 EXE
+head_helper = helper(白名单 python)读前 64 KiB
+head_native = 当前进程读前 64 KiB
+return head_helper != head_native
+```
+
+三种运行环境全部正确：
+
+| 环境 | 加密文件 | 未加密文件 |
+|------|----------|------------|
+| Python / pyd（白名单） | 阶段一即判定 True | 阶段二两视图一致 → False |
+| 打包 EXE（非白名单） | 阶段一相等（都看到密文）→ 阶段二 helper 明文头 ≠ 本地密文头 → True | 阶段二一致 → False |
+
+代价：未加密文件需一次完整 certutil 哈希 + 一次本地完整读取 + helper 前 64 KiB
+读取；大文件应缓存结果。库故意不引入「期望 magic 表」——内容对比与文件类型无关。
+
+## 六、未加密写入通道选择逻辑
+
+目标：任何环境下 `dlp_io.open(path, "w"/"a"/"x")` 落盘都是未加密文件。
+
+```
+首次写某后缀时:
+  1. 环境变量 DLP_IO_WRITE_CHANNEL=direct|powershell → 强制, 跳过探测
+  2. 直写探测: temp 目录写探针文件(同后缀, 含 0x00/0x1A/0xFF 等边界字节)
+     → certutil raw 哈希 == 内容哈希 ?
+     - 一致 → DIRECT（EXE 环境恒真; 本机 python 也真）
+     - certutil 不可用 → 视为无 DLP → DIRECT
+  3. 否则 PowerShell 中转探测: 同 payload 经 stdin 管道由 powershell 子进程
+     ([Console]::OpenStandardInput → [IO.File]::Open) 写盘 → 再校验 raw 哈希
+     - 一致 → POWERSHELL
+  4. 都失败 → raise DlpWriteError（绝不静默写出密文）
+结果按后缀缓存（DLP 写策略常按文档类型区分）。
+```
+
+中转通道要点：字节只走 stdin 内存管道，不落地任何临时文件（白名单进程写的临时
+文件本身可能被加密，会造成「密文套娃」）；`FileMode` 映射 `w→Create`、
+`a→Append`、`x→CreateNew`；`x` 模式在通道选择前先做本地存在性检查，保证
+`FileExistsError` 语义与原生一致。
+
+已知边界：探测文件写在 temp 目录，若 DLP 策略按目录区分，探针结论可能与目标
+目录不符——此时用 `DLP_IO_WRITE_CHANNEL` 显式指定。
+
+## 七、相关已知问题
+
+- 保护区内 `git push` 必现 `fatal: not a git repository`（DLP 拦截 push 进程
+  对 `.git` 的访问模式），解法见 `.kimi-code/skills/git-push-dlp/`（robocopy
+  镜像 `.git` 到 TEMP 后从 stage 目录推送）。
+- 提交样本 fixture 时注意：本机 `git.exe` 是白名单视图，commit 会把加密样本按
+  明文存入仓库。测试用例已按环境自适应编写（依据本机 raw 视图与本地视图是否
+  一致判断期望），CI（无 DLP）与 DLP 开发机都能通过。
@@ -1,17 +1,19 @@
 # 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.3.0`，distribution 名是 `dlp-io`，import 名是 `dlp_io`。
+
+DLP 环境的实测特征（白名单矩阵、密文样本字节对比、各写入候选通道的可用性实验）与判定/写入逻辑的完整推导记录见 [DLP_ENVIRONMENT_NOTES.md](DLP_ENVIRONMENT_NOTES.md)。

 ## 安装

 ### 方式一：Gitea PyPI registry（推荐）

-registry 已开放匿名下载，一条命令即可安装；pip 会自动选择最匹配的 wheel：Windows 上 CPython 3.10–3.14 会得到对应的 pyd wheel（整个库编译为单个 `.pyd`），其余环境得到纯 Python wheel。
+registry 已开放匿名下载，一条命令即可安装；pip 会自动选择最匹配的 wheel：Windows 上 CPython 3.8–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.3.0
 ```

 Package 页面：<https://gitea.docker.antior.cn/antior/-/packages/pypi/dlp-io>
@@ -24,10 +26,10 @@ Release 页面：<https://gitea.docker.antior.cn/antior/dlp-io/releases>

 每个版本包含以下文件：

- `dlp_io-X.Y.Z-py3-none-any.whl`：纯 Python wheel，适用于任何 CPython >= 3.10 环境。
- `dlp_io-X.Y.Z-cp310-cp310-win_amd64.whl` 至 `cp314`：pyd wheel，Windows 专用，按解释器版本区分；整个库由 Cython 编译为单个二进制 `.pyd`，功能与纯 Python 版完全一致，可按对源码保护的要求选用。
+- `dlp_io-X.Y.Z-py3-none-any.whl`：纯 Python wheel，适用于任何 CPython >= 3.8 环境。
+- `dlp_io-X.Y.Z-cp38-cp38-win_amd64.whl` 至 `cp314`：pyd wheel，Windows 专用，按解释器版本区分；整个库由 Cython 编译为单个二进制 `.pyd`，功能与纯 Python 版完全一致，可按对源码保护的要求选用。
 - `dlp_io-X.Y.Z.tar.gz`：sdist 源码包。
- `dlp_io.py` 与 `dlp_io.cp310-win_amd64.pyd` 至 `cp314`：免安装单文件，与 wheel 内容一致；直接放进项目目录（或加入 `PYTHONPATH`）即可 `import dlp_io`，适合不便使用 pip 的环境。
+- `dlp_io.py` 与 `dlp_io.cp38-win_amd64.pyd` 至 `cp314`：免安装单文件，与 wheel 内容一致；直接放进项目目录（或加入 `PYTHONPATH`）即可 `import dlp_io`，适合不便使用 pip 的环境。

 按解释器版本选择对应资产，可直接用 pip 安装 URL。Windows PowerShell：

@@ -78,14 +80,50 @@ dlp_io.shutdown()

 模式路由规则：

- 路径型 `r`、`rt`、`rb`：通过 helper 读取。
- 路径型 `w`、`a`、`x`：使用原生 `open` 写入。
+- 路径型 `r`、`rt`、`rb`：通过 helper 读取，始终返回明文。
+- 路径型 `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 +192,7 @@ finally:
 - `DlpProtocolError`：协议版本、控制消息或响应结构无效。
 - `DlpTransportError`：helper 在显式 EOF 前断开，或文件传输被截断。
 - `DlpSessionBusyError`：同一个 session 已有活动 stream。
+- `DlpWriteError`：没有任何写入通道能落盘未加密文件，或 PowerShell 中转写入失败。

 helper 启动或读取失败时，库会显式报错并且不回退到当前 EXE 直接读取。DLP 直读可能返回合法长度的密文；静默回退会把数据损坏伪装成成功。

@@ -7,11 +7,13 @@ name = "dlp-io"
 dynamic = ["version"]
 description = "io-compatible file reads through an approved Python helper on Windows DLP hosts"
 readme = "docs/DLP_IO_LIBRARY.md"
-requires-python = ">=3.10"
+requires-python = ">=3.8"
 classifiers = [
    "Development Status :: 3 - Alpha",
    "Operating System :: Microsoft :: Windows",
    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
    "Programming Language :: Python :: 3.10",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
@@ -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"
@@ -42,7 +42,7 @@ class PackageClassifier(Enum):

 class WorkflowStep(Enum):
    CLONE_TAG = "Clone immutable package tag"
-    TEST_MATRIX = "Test Python 3.10-3.14"
+    TEST_MATRIX = "Test Python 3.8-3.14"
    BUILD = "Build and check distribution"
    SMOKE = "Smoke test release artifacts"
    RELEASE = "Create Gitea release with wheels"
@@ -108,7 +108,7 @@ def test_pyproject_declares_structured_public_package_contract() -> None:
    project = config["project"]

    assert PackageName(canonicalize_name(project["name"])) is PackageName.DLP_IO
-    assert SpecifierSet(project["requires-python"]) == SpecifierSet(">=3.10")
+    assert SpecifierSet(project["requires-python"]) == SpecifierSet(">=3.8")
    assert {DynamicField(item) for item in project["dynamic"]} == {
        DynamicField.VERSION
    }
Author	SHA1	Message	Date
p40000043244@byd.com	d34861be93	feat: 支持 Python 3.8+ (0.3.0) publish-dlp-io / publish (push) Failing after 2s Details - requires-python >=3.8, classifiers 增加 3.8/3.9 - 代码经 ast feature_version=(3,8) 校验, 无 3.9+ 专属语法; str\|None 等注解由 future annotations 延迟求值 - CI 测试/构建/smoke 矩阵扩展为 3.8-3.14, 裸资产数 6→8 - 注意: Windows runner 需预装 Python 3.8 与 3.9 (py launcher)	2026-08-03 15:30:33 +08:00
p40000043244@byd.com	457e4eed3f	docs: DLP 环境特征调研记录与判定/写入逻辑说明 - docs/DLP_ENVIRONMENT_NOTES.md: 白名单矩阵/密文样本/属性不可区分等实测结论与探索过程 - README: DLP 程序化特征与 is_encrypted/写入通道实现逻辑摘要	2026-07-31 18:32:14 +08:00
p40000043244@byd.com	7b62c23e1e	feat: is_encrypted 加密状态判断与未加密写入通道 (0.2.0) publish-dlp-io / publish (push) Successful in 1m16s Details - is_encrypted(): certutil 原始哈希 + helper 视图对比, Python/pyd/EXE 三环境均可判定 - w/a/x 写入按后缀探测通道, 直写会被加密时自动切换 PowerShell stdin 中转, 保证落盘未加密 - 新增 WriteChannel/DlpWriteError 与 DLP_IO_WRITE_CHANNEL 强制通道开关 - tests: DLP 加解密样本 fixture 与 19 个结构化断言用例	2026-07-31 18:24:22 +08:00
p40000043244@byd.com	a458270fa7	chore: git_push_dlp 改 robocopy 增量镜像（固定 stage 复用，只复制变动文件）	2026-07-31 18:13:44 +08:00
p40000043244@byd.com	e0f1db75c2	docs: 收编 git-push-dlp skill（DLP 保护区内 git push 流程）	2026-07-31 17:59:19 +08:00
antior	65b277fc72	Add PyPI registry install instructions to README	2026-07-31 17:29:40 +08:00
antior	0895501234	Skip existing release assets instead of replacing them	2026-07-31 17:14:40 +08:00
antior	7575a2f807	Use CI_PACKAGE_TOKEN for PyPI registry uploads	2026-07-31 17:04:01 +08:00