From 3b0c81fd9d2664504f24d773d36d710c0156cf47 Mon Sep 17 00:00:00 2001 From: antior Date: Fri, 31 Jul 2026 14:58:37 +0800 Subject: [PATCH] Initial commit: split dlp-io library out of DataPacker --- .gitattributes | 2 + .gitea/workflows/publish-dlp-io.yaml | 453 +++++++++++++++++++++++++++ .gitignore | 42 +++ AGENTS.md | 32 ++ MANIFEST.in | 6 + README.md | 20 ++ dlp_io/__init__.py | 31 ++ dlp_io/_api.py | 144 +++++++++ dlp_io/_config.py | 14 + dlp_io/_errors.py | 22 ++ dlp_io/_helper.py | 172 ++++++++++ dlp_io/_patch.py | 94 ++++++ dlp_io/_raw.py | 134 ++++++++ dlp_io/_session.py | 319 +++++++++++++++++++ dlp_io/py.typed | 0 docs/DLP_IO_LIBRARY.md | 145 +++++++++ pyproject.toml | 33 ++ pytest.ini | 4 + requirements-dev.txt | 7 + tests/test_dlp_io_api.py | 280 +++++++++++++++++ tests/test_dlp_io_packaging.py | 147 +++++++++ tests/test_dlp_io_session.py | 408 ++++++++++++++++++++++++ 22 files changed, 2509 insertions(+) create mode 100644 .gitattributes create mode 100644 .gitea/workflows/publish-dlp-io.yaml create mode 100644 .gitignore create mode 100644 AGENTS.md create mode 100644 MANIFEST.in create mode 100644 README.md create mode 100644 dlp_io/__init__.py create mode 100644 dlp_io/_api.py create mode 100644 dlp_io/_config.py create mode 100644 dlp_io/_errors.py create mode 100644 dlp_io/_helper.py create mode 100644 dlp_io/_patch.py create mode 100644 dlp_io/_raw.py create mode 100644 dlp_io/_session.py create mode 100644 dlp_io/py.typed create mode 100644 docs/DLP_IO_LIBRARY.md create mode 100644 pyproject.toml create mode 100644 pytest.ini create mode 100644 requirements-dev.txt create mode 100644 tests/test_dlp_io_api.py create mode 100644 tests/test_dlp_io_packaging.py create mode 100644 tests/test_dlp_io_session.py diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..2b65f6f --- /dev/null +++ b/.gitattributes @@ -0,0 +1,2 @@ +*.bat text eol=crlf +*.cmd text eol=crlf diff --git a/.gitea/workflows/publish-dlp-io.yaml b/.gitea/workflows/publish-dlp-io.yaml new file mode 100644 index 0000000..a905776 --- /dev/null +++ b/.gitea/workflows/publish-dlp-io.yaml @@ -0,0 +1,453 @@ +name: publish-dlp-io + +# Publish the standalone dlp-io distribution from an immutable dlp-io-vX.Y.Z tag. +# The Windows runner cannot reach github.com, so this workflow uses no third-party actions. + +on: + push: + tags: + - "dlp-io-v*.*.*" + workflow_dispatch: + inputs: + version: + description: "Immutable package tag, for example dlp-io-v0.1.0" + required: true + type: string + +env: + PIP_INDEX_URL: https://pypi.tuna.tsinghua.edu.cn/simple + +jobs: + publish: + runs-on: windows-latest + defaults: + run: + working-directory: repo + steps: + - name: Clone immutable package tag + shell: pwsh + working-directory: . + env: + ACTION_EVENT_NAME: ${{ github.event_name }} + ACTION_REF_NAME: ${{ github.ref_name }} + ACTION_SERVER_URL: ${{ github.server_url }} + ACTION_REPOSITORY: ${{ github.repository }} + ACTION_CLONE_TOKEN: ${{ secrets.GITHUB_TOKEN }} + DLP_IO_INPUT_VERSION: ${{ inputs.version }} + run: | + $ErrorActionPreference = "Stop" + $tag = if ($env:ACTION_EVENT_NAME -eq "workflow_dispatch") { + $env:DLP_IO_INPUT_VERSION + } else { + $env:ACTION_REF_NAME + } + if ($tag -notmatch '^dlp-io-v\d+\.\d+\.\d+$') { + throw "Package tag must use dlp-io-vMAJOR.MINOR.PATCH format: $tag" + } + $packageVersion = $tag.Substring("dlp-io-v".Length) + + $token = $env:ACTION_CLONE_TOKEN + if ([string]::IsNullOrWhiteSpace($token)) { + throw "GITHUB_TOKEN is required to clone the repository" + } + $server = [Uri]$env:ACTION_SERVER_URL + $clone = "$($server.Scheme)://x-access-token:$token@$($server.Authority)/$($env:ACTION_REPOSITORY).git" + $masked = [regex]::Escape($token) + $tagRef = "refs/tags/$tag" + $peeledRef = "$tagRef^{}" + $remoteOutput = @(git ls-remote --tags $clone $tagRef $peeledRef 2>&1) + $remoteExit = $LASTEXITCODE + $remoteOutput | ForEach-Object { $_ -replace $masked, "***" } + if ($remoteExit -ne 0) { throw "remote tag lookup failed: $remoteExit" } + $remoteRefs = @{} + foreach ($line in $remoteOutput) { + if ($line -match '^([0-9a-f]{40})\s+(.+)$') { + $remoteRefs[$Matches[2]] = $Matches[1] + } + } + if (-not $remoteRefs.ContainsKey($tagRef)) { + throw "Remote package tag does not exist: $tagRef" + } + if (-not $remoteRefs.ContainsKey($peeledRef)) { + throw "Package tag must be annotated: $tagRef" + } + $remoteCommit = $remoteRefs[$peeledRef] + + git clone -c core.autocrlf=true --depth 1 --branch $tag $clone repo 2>&1 | + ForEach-Object { $_ -replace $masked, "***" } + if ($LASTEXITCODE -ne 0) { throw "tagged source clone failed: $LASTEXITCODE" } + + Push-Location repo + try { + $sourceVersion = (& py -3.13 -c "import dlp_io; print(dlp_io.__version__)").Trim() + } finally { + Pop-Location + } + if ($sourceVersion -ne $packageVersion) { + throw "dlp-io tag/version mismatch: tag=$tag package=$sourceVersion" + } + $tagCommit = (git -C repo rev-list -n 1 $tag).Trim() + $headCommit = (git -C repo rev-parse HEAD).Trim() + if ($tagCommit -ne $remoteCommit -or $remoteCommit -ne $headCommit) { + throw "Immutable tag checkout mismatch: remote=$remoteCommit local=$tagCommit HEAD=$headCommit" + } + "DLP_IO_PACKAGE_TAG=$tag" | Out-File $env:GITHUB_ENV -Encoding utf8 -Append + "DLP_IO_PACKAGE_VERSION=$packageVersion" | Out-File $env:GITHUB_ENV -Encoding utf8 -Append + Write-Host "package source: $tag @ $headCommit" + + - name: Test Python 3.9-3.13 + shell: pwsh + run: | + $ErrorActionPreference = "Stop" + foreach ($pythonVersion in @("3.9", "3.10", "3.11", "3.12", "3.13")) { + & py "-$pythonVersion" --version + if ($LASTEXITCODE -ne 0) { throw "Python $pythonVersion is required" } + & py "-$pythonVersion" -m pip install -r requirements-dev.txt --progress-bar off + if ($LASTEXITCODE -ne 0) { throw "test dependency install failed on $pythonVersion" } + $abi = $pythonVersion.Replace(".", "") + $baseTemp = Join-Path $env:TEMP "dlp-io-py$abi-$PID" + & py "-$pythonVersion" -m pytest -q --basetemp $baseTemp + if ($LASTEXITCODE -ne 0) { throw "full pytest failed on $pythonVersion" } + } + + - name: Build and check distribution + shell: pwsh + run: | + $ErrorActionPreference = "Stop" + py -3.13 -m pip install build twine --progress-bar off + if ($LASTEXITCODE -ne 0) { throw "build dependency install failed" } + py -3.13 -m build + if ($LASTEXITCODE -ne 0) { throw "dlp-io build failed" } + py -3.13 -m twine check "dist\dlp_io-$($env:DLP_IO_PACKAGE_VERSION)*" + if ($LASTEXITCODE -ne 0) { throw "twine check failed" } + + $version = $env:DLP_IO_PACKAGE_VERSION + $expected = @( + "dlp_io-$version-py3-none-any.whl", + "dlp_io-$version.tar.gz" + ) + $actual = @(Get-ChildItem -LiteralPath dist -File | ForEach-Object Name) + if (Compare-Object $expected $actual) { + throw "distribution file set mismatch: $($actual -join ', ')" + } + + - name: Smoke test isolated wheel + shell: pwsh + run: | + $ErrorActionPreference = "Stop" + $root = Join-Path $env:TEMP ("dlp-io-wheel-smoke-" + $PID) + $rootFull = [IO.Path]::GetFullPath($root) + $tempFull = [IO.Path]::GetFullPath($env:TEMP).TrimEnd('\') + '\' + if (-not $rootFull.StartsWith($tempFull, [StringComparison]::OrdinalIgnoreCase)) { + throw "unsafe wheel smoke path: $rootFull" + } + try { + py -3.13 -m venv $rootFull + if ($LASTEXITCODE -ne 0) { throw "wheel smoke venv creation failed" } + $python = Join-Path $rootFull "Scripts\python.exe" + $dist = (Resolve-Path dist).Path + & $python -m pip install --no-index --find-links $dist "dlp-io==$env:DLP_IO_PACKAGE_VERSION" --progress-bar off + if ($LASTEXITCODE -ne 0) { throw "wheel installation failed" } + Push-Location $rootFull + try { + & $python -c "import dlp_io,sys,tempfile; from pathlib import Path; assert dlp_io.__version__ == '$env:DLP_IO_PACKAGE_VERSION'; td=tempfile.TemporaryDirectory(); p=Path(td.name)/'payload.bin'; w=dlp_io.open(p,'wb'); assert w.write(b'0123456789') == 10; w.close(); dlp_io.configure(python_executable=sys.executable); r=dlp_io.open(p,'rb'); assert r.read(4) == b'0123'; assert r.seek(-3,2) == 7; assert r.read() == b'789'; r.close(); dlp_io.shutdown(); td.cleanup(); print('wheel smoke passed')" + if ($LASTEXITCODE -ne 0) { throw "wheel smoke failed" } + } finally { + Pop-Location + } + } finally { + if (Test-Path -LiteralPath $rootFull) { + Remove-Item -LiteralPath $rootFull -Recurse -Force + } + } + + - name: Upload package to Gitea PyPI + shell: pwsh + env: + TWINE_USERNAME: antior + TWINE_PASSWORD: ${{ secrets.CI_PACKAGE_TOKEN }} + TWINE_REPOSITORY_URL: https://gitea.docker.antior.cn/api/packages/antior/pypi + run: | + $ErrorActionPreference = "Stop" + if ([string]::IsNullOrWhiteSpace($env:TWINE_PASSWORD)) { + throw "CI_PACKAGE_TOKEN repository secret is required" + } + + function Get-RegistryEntries([string]$SimpleUrl, [hashtable]$Headers) { + try { + $response = Invoke-WebRequest -Headers $Headers -Uri $SimpleUrl + } catch { + if ([int]$_.Exception.Response.StatusCode -eq 404) { return @() } + throw + } + $pattern = '[^"]+)"[^>]*>(?[^<]+)' + foreach ($match in [regex]::Matches($response.Content, $pattern, "IgnoreCase")) { + $href = [Net.WebUtility]::HtmlDecode($match.Groups["href"].Value) + $uri = [Uri]$href + $hashMatch = [regex]::Match($uri.Fragment, '^#sha256=(?[0-9a-fA-F]{64})$') + if (-not $uri.IsAbsoluteUri -or -not $hashMatch.Success) { + throw "registry simple index contains an invalid artifact link" + } + $builder = [UriBuilder]$uri + $builder.Fragment = "" + [pscustomobject]@{ + filename = [Net.WebUtility]::HtmlDecode($match.Groups["name"].Value) + url = $builder.Uri.AbsoluteUri + sha256 = $hashMatch.Groups["hash"].Value.ToLowerInvariant() + } + } + } + + function Get-ArchiveContentDigest([string]$Path) { + $script = @' + import hashlib + import sys + import tarfile + import zipfile + from pathlib import Path, PurePosixPath + + path = Path(sys.argv[1]) + entries = [] + if path.suffix == ".whl": + with zipfile.ZipFile(path) as archive: + entries = [ + (item.filename, archive.read(item)) + for item in archive.infolist() + if not item.is_dir() + ] + else: + with tarfile.open(path, "r:*") as archive: + for item in archive.getmembers(): + if not item.isfile(): + continue + stream = archive.extractfile(item) + parts = PurePosixPath(item.name).parts[1:] + if stream is not None and parts: + entries.append(("/".join(parts), stream.read())) + + digest = hashlib.sha256() + for name, content in sorted(entries): + digest.update(name.encode("utf-8")) + digest.update(b"\0") + digest.update(len(content).to_bytes(8, "big")) + digest.update(hashlib.sha256(content).digest()) + print(digest.hexdigest()) + '@ + $digest = (& py -3.13 -c $script $Path).Trim() + if ($LASTEXITCODE -ne 0 -or $digest -notmatch '^[0-9a-f]{64}$') { + throw "failed to calculate archive content digest: $Path" + } + return $digest + } + + $packageVersion = $env:DLP_IO_PACKAGE_VERSION + $expected = @( + "dlp_io-$packageVersion-py3-none-any.whl", + "dlp_io-$packageVersion.tar.gz" + ) + $headers = @{ Authorization = "token $env:TWINE_PASSWORD" } + $simpleUrl = "$env:TWINE_REPOSITORY_URL/simple/dlp-io/" + $registryOrigin = [Uri]$env:TWINE_REPOSITORY_URL + $entries = @(Get-RegistryEntries $simpleUrl $headers) + $missingArtifacts = [System.Collections.Generic.List[string]]::new() + $root = Join-Path $env:TEMP ("dlp-io-upload-state-" + [Guid]::NewGuid().ToString("N")) + $rootFull = [IO.Path]::GetFullPath($root) + $tempFull = [IO.Path]::GetFullPath($env:TEMP).TrimEnd('\') + '\' + if (-not $rootFull.StartsWith($tempFull, [StringComparison]::OrdinalIgnoreCase)) { + throw "unsafe upload verification path: $rootFull" + } + try { + New-Item -ItemType Directory -Force $rootFull | Out-Null + foreach ($filename in $expected) { + $local = (Resolve-Path (Join-Path dist $filename)).Path + $entry = @($entries | Where-Object filename -eq $filename) + if ($entry.Count -eq 0) { + $missingArtifacts.Add($local) + continue + } + if ($entry.Count -ne 1) { + throw "registry simple index contains duplicate artifact links: $filename" + } + $downloadUri = [Uri]$entry[0].url + if (-not $downloadUri.Scheme.Equals($registryOrigin.Scheme, [StringComparison]::OrdinalIgnoreCase) -or + -not $downloadUri.Authority.Equals($registryOrigin.Authority, [StringComparison]::OrdinalIgnoreCase)) { + throw "download URL is not on the package registry origin: $filename" + } + $remote = Join-Path $rootFull $filename + Invoke-WebRequest -Headers $headers -Uri $downloadUri.AbsoluteUri -OutFile $remote + $remoteHash = (Get-FileHash -LiteralPath $remote -Algorithm SHA256).Hash.ToLowerInvariant() + if ($remoteHash -ne $entry[0].sha256) { + throw "registry download hash mismatch: $filename" + } + if ((Get-ArchiveContentDigest $local) -ne (Get-ArchiveContentDigest $remote)) { + throw "existing registry artifact differs from tagged build: $filename" + } + Write-Host "registry artifact already matches tagged build: $filename" + } + + if ($missingArtifacts.Count -gt 0) { + $twineArgs = @( + "-3.13", "-m", "twine", "upload", "--non-interactive", + "--disable-progress-bar", "--repository-url", $env:TWINE_REPOSITORY_URL + ) + $missingArtifacts + & py @twineArgs + if ($LASTEXITCODE -ne 0) { throw "twine upload failed" } + } else { + Write-Host "all registry artifacts already match the tagged build" + } + } finally { + if (Test-Path -LiteralPath $rootFull) { + Remove-Item -LiteralPath $rootFull -Recurse -Force + } + } + + - name: Test registry package files + shell: pwsh + env: + REGISTRY_TOKEN: ${{ secrets.CI_PACKAGE_TOKEN }} + TWINE_REPOSITORY_URL: https://gitea.docker.antior.cn/api/packages/antior/pypi + run: | + $ErrorActionPreference = "Stop" + $packageVersion = $env:DLP_IO_PACKAGE_VERSION + $headers = @{ Authorization = "token $env:REGISTRY_TOKEN" } + + function Get-RegistryEntries([string]$SimpleUrl, [hashtable]$Headers) { + try { + $response = Invoke-WebRequest -Headers $Headers -Uri $SimpleUrl + } catch { + if ([int]$_.Exception.Response.StatusCode -eq 404) { return @() } + throw + } + $pattern = '[^"]+)"[^>]*>(?[^<]+)' + foreach ($match in [regex]::Matches($response.Content, $pattern, "IgnoreCase")) { + $href = [Net.WebUtility]::HtmlDecode($match.Groups["href"].Value) + $uri = [Uri]$href + $hashMatch = [regex]::Match($uri.Fragment, '^#sha256=(?[0-9a-fA-F]{64})$') + if (-not $uri.IsAbsoluteUri -or -not $hashMatch.Success) { + throw "registry simple index contains an invalid artifact link" + } + $builder = [UriBuilder]$uri + $builder.Fragment = "" + [pscustomobject]@{ + filename = [Net.WebUtility]::HtmlDecode($match.Groups["name"].Value) + url = $builder.Uri.AbsoluteUri + sha256 = $hashMatch.Groups["hash"].Value.ToLowerInvariant() + } + } + } + + function Get-ArchiveContentDigest([string]$Path) { + $script = @' + import hashlib + import sys + import tarfile + import zipfile + from pathlib import Path, PurePosixPath + + path = Path(sys.argv[1]) + entries = [] + if path.suffix == ".whl": + with zipfile.ZipFile(path) as archive: + entries = [ + (item.filename, archive.read(item)) + for item in archive.infolist() + if not item.is_dir() + ] + else: + with tarfile.open(path, "r:*") as archive: + for item in archive.getmembers(): + if not item.isfile(): + continue + stream = archive.extractfile(item) + parts = PurePosixPath(item.name).parts[1:] + if stream is not None and parts: + entries.append(("/".join(parts), stream.read())) + + digest = hashlib.sha256() + for name, content in sorted(entries): + digest.update(name.encode("utf-8")) + digest.update(b"\0") + digest.update(len(content).to_bytes(8, "big")) + digest.update(hashlib.sha256(content).digest()) + print(digest.hexdigest()) + '@ + $digest = (& py -3.13 -c $script $Path).Trim() + if ($LASTEXITCODE -ne 0 -or $digest -notmatch '^[0-9a-f]{64}$') { + throw "failed to calculate archive content digest: $Path" + } + return $digest + } + + $expected = @( + "dlp_io-$packageVersion-py3-none-any.whl", + "dlp_io-$packageVersion.tar.gz" + ) + $simpleUrl = "$env:TWINE_REPOSITORY_URL/simple/dlp-io/" + $entries = @() + $actual = @() + for ($attempt = 1; $attempt -le 20; $attempt++) { + $entries = @(Get-RegistryEntries $simpleUrl $headers) + $actual = @($entries | ForEach-Object filename) + if (-not (Compare-Object $expected $actual)) { break } + if ($attempt -lt 20) { Start-Sleep -Seconds 3 } + } + if (Compare-Object $expected $actual) { + throw "registry package file mismatch: $($actual -join ', ')" + } + + $registryRoot = Join-Path $env:TEMP ("dlp-io-registry-smoke-" + $PID) + $registryRootFull = [IO.Path]::GetFullPath($registryRoot) + $tempFull = [IO.Path]::GetFullPath($env:TEMP).TrimEnd('\') + '\' + if (-not $registryRootFull.StartsWith($tempFull, [StringComparison]::OrdinalIgnoreCase)) { + throw "unsafe registry smoke path: $registryRootFull" + } + try { + $downloads = Join-Path $registryRootFull "downloads" + New-Item -ItemType Directory -Force $downloads | Out-Null + $registryOrigin = [Uri]$env:TWINE_REPOSITORY_URL + foreach ($filename in $expected) { + $entry = @($entries | Where-Object filename -eq $filename) + if ($entry.Count -ne 1 -or [string]::IsNullOrWhiteSpace($entry[0].url)) { + throw "registry simple index is missing a download URL: $filename" + } + if ([string]::IsNullOrWhiteSpace($entry[0].sha256)) { + throw "registry simple index is missing SHA-256: $filename" + } + $downloadUri = [Uri]$entry[0].url + if (-not $downloadUri.IsAbsoluteUri -or + -not $downloadUri.Scheme.Equals($registryOrigin.Scheme, [StringComparison]::OrdinalIgnoreCase) -or + -not $downloadUri.Authority.Equals($registryOrigin.Authority, [StringComparison]::OrdinalIgnoreCase)) { + throw "download URL is not on the package registry origin: $filename" + } + $target = Join-Path $downloads $filename + Invoke-WebRequest -Headers $headers -Uri $downloadUri.AbsoluteUri -OutFile $target + $downloadHash = (Get-FileHash -LiteralPath $target -Algorithm SHA256).Hash.ToLowerInvariant() + if ($downloadHash -ne $entry[0].sha256) { + throw "registry download hash mismatch: $filename" + } + $local = (Resolve-Path (Join-Path dist $filename)).Path + if ((Get-ArchiveContentDigest $local) -ne (Get-ArchiveContentDigest $target)) { + throw "registry artifact differs from tagged build: $filename" + } + } + + $venv = Join-Path $registryRootFull "venv" + py -3.13 -m venv $venv + if ($LASTEXITCODE -ne 0) { throw "registry smoke venv creation failed" } + $python = Join-Path $venv "Scripts\python.exe" + $registryWheel = Join-Path $downloads "dlp_io-$packageVersion-py3-none-any.whl" + & $python -m pip install --no-index --no-deps $registryWheel --progress-bar off + if ($LASTEXITCODE -ne 0) { throw "registry package installation failed" } + Push-Location $registryRootFull + try { + & $python -c "import dlp_io,sys,tempfile; from pathlib import Path; assert dlp_io.__version__ == '$packageVersion'; td=tempfile.TemporaryDirectory(); p=Path(td.name)/'payload.bin'; w=dlp_io.open(p,'wb'); w.write(b'registry'); w.close(); dlp_io.configure(python_executable=sys.executable); r=dlp_io.open(p,'rb'); assert r.read() == b'registry'; r.close(); dlp_io.shutdown(); td.cleanup(); print('registry install passed')" + if ($LASTEXITCODE -ne 0) { throw "registry installation smoke failed" } + } finally { + Pop-Location + } + } finally { + if (Test-Path -LiteralPath $registryRootFull) { + Remove-Item -LiteralPath $registryRootFull -Recurse -Force + } + } + Write-Host "registry verified: dlp-io==$packageVersion ($($actual.Count) files)" diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ed56903 --- /dev/null +++ b/.gitignore @@ -0,0 +1,42 @@ +# Python cache +__pycache__/ +**/__pycache__/ +*.py[cod] +*.pyc +*$py.class + +# Distribution / packaging +build/ +dist/ +*.egg-info/ +*.egg + +# Virtual environments +.env +.venv/ +venv/ +ENV/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo + +# OS files +.DS_Store +Thumbs.db +Desktop.ini + +# Logs +*.log + +# Test output +.pytest_cache/ +.coverage +htmlcov/ + +# Temporary files +*.tmp +*.temp +*.bak diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..4ba0fcf --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,32 @@ +# DataPacker 项目 Agent 约束 + +## 严禁文本断言（Strict Ban on Text Assertions） + +编写或修改 pytest 测试时，绝对禁止任何形式的纯文本或字符串断言。 + +### 禁止写法 + +- 禁止 `assert "some string" in result`。 +- 禁止 `assert result == "exact string"`。 +- 禁止 `assert "string" not in result`。 +- 禁止使用正则表达式匹配纯文本输出。 +- 禁止使用 `pytest.raises(..., match=...)` 校验异常消息。 +- 禁止通过读取 Markdown、源码、脚本、workflow 或配置文件后搜索字符串来验证行为。 + +### 必须采用的验证方式 + +- 验证数据结构或对象状态：断言函数返回的字典、Dataclass、Pydantic Model、Enum、Path 或其他领域对象属性，不断言其序列化文本。 +- 验证方法调用：使用 `unittest.mock.patch`、`MagicMock` 或 spy，断言底层核心函数收到的参数和调用次数。 +- 验证业务副产物：断言文件、archive members、hash、状态码、布尔值、数值、对象类型或进程退出状态。 +- 验证结构化配置：TOML、YAML、JSON 必须解析成对象后验证；不得搜索原始文本。 +- 验证异常时优先断言异常类型和结构化属性，不得匹配异常消息。 + +### 被测代码只有文本输出时 + +如果被测函数只返回一个巨大字符串，必须先重构业务代码，把“数据组装”和“文本渲染”拆开；测试只覆盖数据组装函数返回的结构化对象。不得为了保留文本断言而增加另一种字符串搜索、正则或快照测试。 + +### 项目门禁 + +- 项目内所有 `test_*.py`、`*_test.py` 和 `conftest.py` 都必须通过文本断言 AST policy gate，包括根测试集之外的子项目测试。 +- 新测试和既有测试适用同一规则，不允许 grandfathered violation。 +- 字符串可以作为被测 API 输入、Path 构造参数、字典 key 或 mock 调用参数；禁止的是对纯文本结果、渲染文本、源码文本、日志文本和异常消息进行断言。 diff --git a/MANIFEST.in b/MANIFEST.in new file mode 100644 index 0000000..3b000f1 --- /dev/null +++ b/MANIFEST.in @@ -0,0 +1,6 @@ +include pyproject.toml +include docs/DLP_IO_LIBRARY.md +recursive-include dlp_io *.py py.typed +prune tests +exclude README.md +global-exclude __pycache__ *.py[cod] .env diff --git a/README.md b/README.md new file mode 100644 index 0000000..1c0adfc --- /dev/null +++ b/README.md @@ -0,0 +1,20 @@ +# dlp-io + +io-compatible file reads through an approved Python helper on Windows DLP hosts. + +完整文档见 [docs/DLP_IO_LIBRARY.md](docs/DLP_IO_LIBRARY.md)。 + +## 开发 + +```bash +pip install -r requirements-dev.txt +pytest +``` + +## 构建发布 + +```bash +py -3.13 -m build +``` + +发布流程由 `.gitea/workflows/publish-dlp-io.yaml` 驱动：推送 `dlp-io-vX.Y.Z` 格式的 annotated tag 即可触发测试、构建、上传到 Gitea PyPI 仓库。 diff --git a/dlp_io/__init__.py b/dlp_io/__init__.py new file mode 100644 index 0000000..3eb4a38 --- /dev/null +++ b/dlp_io/__init__.py @@ -0,0 +1,31 @@ +from ._api import configure, open, shutdown +from ._config import DlpConfig +from ._errors import ( + DlpConfigurationError, + DlpHelperStartError, + DlpIoError, + DlpProtocolError, + DlpSessionBusyError, + DlpTransportError, +) +from ._session import DlpSession +from ._patch import install_open_patch, install_reader_patch, uninstall_open_patch + +__version__ = "0.1.0" + +__all__ = [ + "DlpConfig", + "DlpConfigurationError", + "DlpHelperStartError", + "DlpIoError", + "DlpProtocolError", + "DlpSession", + "DlpSessionBusyError", + "DlpTransportError", + "configure", + "install_open_patch", + "install_reader_patch", + "open", + "shutdown", + "uninstall_open_patch", +] diff --git a/dlp_io/_api.py b/dlp_io/_api.py new file mode 100644 index 0000000..85b10be --- /dev/null +++ b/dlp_io/_api.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +import atexit +import builtins +import io +import operator +import os +import threading + +from ._config import DlpConfig +from ._errors import DlpConfigurationError +from ._session import DlpSession + +_ORIGINAL_OPEN = builtins.open +_DEFAULT_BUFFER_SIZE = 1024 * 1024 +_state_lock = threading.RLock() +_config = DlpConfig() +_default_session = None + + +def _validate_path_mode(mode, buffering, encoding, errors, newline) -> int: + buffering = operator.index(buffering) + if buffering < -1: + raise ValueError("invalid buffering size") + if not isinstance(mode, str): + raise TypeError("open() argument 'mode' must be str, not %s" % type(mode).__name__) + allowed = set("axrwb+t") + invalid = set(mode) - allowed + duplicate = any(mode.count(character) > 1 for character in allowed) + if not mode or invalid or duplicate or sum(c in mode for c in "axrw") != 1: + raise ValueError("invalid mode: %r" % mode) + if "b" in mode and "t" in mode: + raise ValueError("can't have text and binary mode at once") + if "b" in mode: + if encoding is not None: + raise ValueError("binary mode doesn't take an encoding argument") + if errors is not None: + raise ValueError("binary mode doesn't take an errors argument") + if newline is not None: + raise ValueError("binary mode doesn't take a newline argument") + elif buffering == 0: + raise ValueError("can't have unbuffered text I/O") + return buffering + + +def configure(*, python_executable=None, startup_timeout=60) -> None: + """Configure the lazy default session before its first bridged read.""" + global _config + with _state_lock: + if _default_session is not None: + raise DlpConfigurationError( + "cannot configure dlp_io while the default session is active" + ) + if startup_timeout <= 0: + raise DlpConfigurationError("startup_timeout must be positive") + _config = DlpConfig( + python_executable=python_executable, + startup_timeout=startup_timeout, + ) + + +def _get_default_session(): + global _default_session + with _state_lock: + if _default_session is None: + _default_session = DlpSession.start( + python_cmd=_config.python_command(), + timeout=_config.startup_timeout, + ) + return _default_session + + +def _wrap_reader(raw, mode, buffering, encoding, errors, newline): + binary = "b" in mode + if binary and buffering == 0: + return raw + if not binary and buffering == 0: + raw.close() + raise ValueError("can't have unbuffered text I/O") + 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.BufferedReader(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( + file, + mode="r", + buffering=-1, + encoding=None, + errors=None, + newline=None, + closefd=True, + opener=None, +): + """Open a file with io.open-compatible calling conventions.""" + is_path = isinstance(file, (str, bytes, os.PathLike)) + if not is_path or opener is not None: + return _ORIGINAL_OPEN( + file, + mode, + buffering, + encoding, + errors, + newline, + closefd, + opener, + ) + buffering = _validate_path_mode(mode, buffering, encoding, errors, newline) + if "+" in mode: + raise io.UnsupportedOperation( + "dlp_io does not support update mode for path-based files" + ) + pure_read = "r" in mode + if pure_read: + if not closefd: + raise ValueError("Cannot use closefd=False with file name") + 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) + + +def shutdown() -> None: + """Close and forget the lazy default session. This operation is idempotent.""" + global _default_session + with _state_lock: + session = _default_session + _default_session = None + if session is not None: + session.close() + + +atexit.register(shutdown) diff --git a/dlp_io/_config.py b/dlp_io/_config.py new file mode 100644 index 0000000..775609a --- /dev/null +++ b/dlp_io/_config.py @@ -0,0 +1,14 @@ +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class DlpConfig: + python_executable: str | None = None + startup_timeout: float = 60 + + def python_command(self) -> list[str] | None: + if self.python_executable is None: + return None + return [self.python_executable] diff --git a/dlp_io/_errors.py b/dlp_io/_errors.py new file mode 100644 index 0000000..8e9eed7 --- /dev/null +++ b/dlp_io/_errors.py @@ -0,0 +1,22 @@ +class DlpIoError(Exception): + """Base exception for dlp-io failures.""" + + +class DlpConfigurationError(DlpIoError): + """The bridge configuration is invalid for the current lifecycle state.""" + + +class DlpHelperStartError(DlpIoError): + """The approved Python helper could not be started.""" + + +class DlpProtocolError(DlpIoError): + """The helper and client exchanged an invalid protocol message.""" + + +class DlpTransportError(DlpIoError): + """The authenticated pipe stream failed before a valid EOF.""" + + +class DlpSessionBusyError(DlpIoError): + """A session already owns an active file stream.""" diff --git a/dlp_io/_helper.py b/dlp_io/_helper.py new file mode 100644 index 0000000..202c115 --- /dev/null +++ b/dlp_io/_helper.py @@ -0,0 +1,172 @@ +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + +from ._errors import DlpConfigurationError + +PROTOCOL_VERSION = 1 +CONTROL_LIMIT = 64 * 1024 +CHUNK_SIZE = 1024 * 1024 +HANDSHAKE_PREFIX = "DLP_IO 1 " + + +class HelperErrorCode(str, Enum): + INVALID_PATH = "invalid_path" + INVALID_OFFSET = "invalid_offset" + FILE_ACCESS = "file_access" + INVALID_REQUEST = "invalid_request" + INVALID_REQUEST_TYPE = "invalid_request_type" + UNSUPPORTED_VERSION = "unsupported_version" + UNKNOWN_OPERATION = "unknown_operation" + INTERNAL = "internal" + + +@dataclass(frozen=True) +class HelperSourceConfig: + authkey: bytes + protocol_version: int = PROTOCOL_VERSION + control_limit: int = CONTROL_LIMIT + chunk_size: int = CHUNK_SIZE + + @classmethod + def from_authkey(cls, authkey: bytes) -> HelperSourceConfig: + if not isinstance(authkey, bytes) or len(authkey) != 32: + raise DlpConfigurationError("authkey must contain exactly 32 bytes") + return cls(authkey=authkey) + + +_HELPER_TEMPLATE = r''' +import json +import os +import sys +from multiprocessing.connection import AuthenticationError, Listener + +VERSION = __PROTOCOL_VERSION__ +CONTROL_LIMIT = __CONTROL_LIMIT__ +CHUNK_SIZE = __CHUNK_SIZE__ +PIPE_PREFIX = r"\\.\pipe\DlpIo_" +AUTHKEY = bytes.fromhex("__AUTHKEY_HEX__") + + +def send_json(conn, value): + conn.send_bytes(json.dumps(value).encode("utf-8")) + + +def error(conn, code, message): + send_json(conn, { + "version": VERSION, + "ok": False, + "error_code": code, + "error": message, + }) + + +def handle_read(conn, request): + path = request.get("path") + offset = request.get("offset", 0) + if not isinstance(path, str) or not path: + error(conn, "invalid_path", "path must be a non-empty string") + return + if isinstance(offset, bool) or not isinstance(offset, int) or offset < 0: + error(conn, "invalid_offset", "offset must be a non-negative integer") + return + try: + size = os.path.getsize(path) + stream = open(path, "rb") + except Exception as exc: + error(conn, "file_access", "%s: %s" % (type(exc).__name__, exc)) + return + try: + stream.seek(offset) + send_json(conn, {"version": VERSION, "ok": True, "size": size}) + while True: + chunk = stream.read(CHUNK_SIZE) + if not chunk: + conn.send_bytes(b"") + return + conn.send_bytes(chunk) + except (BrokenPipeError, EOFError, OSError): + return + finally: + stream.close() + + +def handle(conn): + try: + request = json.loads(conn.recv_bytes(maxlength=CONTROL_LIMIT).decode("utf-8")) + except Exception as exc: + error(conn, "invalid_request", "invalid request: %s" % exc) + return True + if not isinstance(request, dict): + error(conn, "invalid_request_type", "request must be a JSON object") + return True + if request.get("version") != VERSION: + error(conn, "unsupported_version", "unsupported protocol version") + return True + operation = request.get("op") + if operation == "ping": + send_json(conn, { + "version": VERSION, + "ok": True, + "pid": os.getpid(), + "python": sys.version, + }) + return True + if operation == "read": + handle_read(conn, request) + return True + if operation == "quit": + send_json(conn, {"version": VERSION, "ok": True}) + return False + error(conn, "unknown_operation", "unknown operation: %r" % operation) + return True + + +def main(): + address = PIPE_PREFIX + os.urandom(16).hex() + listener = Listener(address=address, family="AF_PIPE", authkey=AUTHKEY) + sys.stdout.write("DLP_IO " + str(VERSION) + " " + address + "\n") + sys.stdout.flush() + running = True + while running: + try: + conn = listener.accept() + except AuthenticationError: + continue + try: + running = handle(conn) + except Exception as exc: + try: + error(conn, "internal", "%s: %s" % (type(exc).__name__, exc)) + except Exception: + pass + finally: + try: + conn.close() + except Exception: + pass + listener.close() + + +try: + main() +except Exception as exc: + sys.stderr.write("dlp-io helper fatal: %s: %s\n" % (type(exc).__name__, exc)) + sys.stderr.flush() + sys.exit(1) +''' + + +def render_helper_source(authkey: bytes) -> str: + config = HelperSourceConfig.from_authkey(authkey) + replacements = { + "__AUTHKEY_HEX__": config.authkey.hex(), + "__PROTOCOL_VERSION__": str(config.protocol_version), + "__CONTROL_LIMIT__": str(config.control_limit), + "__CHUNK_SIZE__": str(config.chunk_size), + } + source = _HELPER_TEMPLATE + for marker, value in replacements.items(): + source = source.replace(marker, value) + return source diff --git a/dlp_io/_patch.py b/dlp_io/_patch.py new file mode 100644 index 0000000..3f0aa2a --- /dev/null +++ b/dlp_io/_patch.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import builtins +import io +import os +import threading + +from ._api import _ORIGINAL_OPEN, _validate_path_mode, _wrap_reader, open as dlp_open + +_ORIGINAL_IO_OPEN = io.open +_patch_lock = threading.RLock() +_installed = False + + +def _reader_wrapper(open_read_func): + def patched_open( + file, + mode="r", + buffering=-1, + encoding=None, + errors=None, + newline=None, + closefd=True, + opener=None, + ): + is_path = isinstance(file, (str, bytes, os.PathLike)) + if not is_path or opener is not None: + return _ORIGINAL_OPEN( + file, + mode, + buffering, + encoding, + errors, + newline, + closefd, + opener, + ) + buffering = _validate_path_mode(mode, buffering, encoding, errors, newline) + if "+" in mode: + raise io.UnsupportedOperation( + "dlp_io does not support update mode for path-based files" + ) + pure_read = "r" in mode + if pure_read: + if not closefd: + raise ValueError("Cannot use closefd=False with file name") + reader = open_read_func(os.fsdecode(os.fspath(file))) + if isinstance(reader, io.RawIOBase): + return _wrap_reader(reader, mode, buffering, encoding, errors, newline) + if "b" in mode: + return reader + return io.TextIOWrapper( + reader, + encoding=encoding, + errors=errors, + newline=newline, + line_buffering=buffering == 1, + ) + return _ORIGINAL_OPEN(file, mode, buffering, encoding, errors, newline, closefd) + + return patched_open + + +def install_open_patch() -> None: + """Route global path-based reads through the lazy default dlp-io session.""" + global _installed + with _patch_lock: + if _installed: + return + builtins.open = dlp_open + io.open = dlp_open + _installed = True + + +def install_reader_patch(open_read_func) -> None: + """Compatibility hook for an explicitly managed session.open_raw callback.""" + global _installed + with _patch_lock: + if _installed: + return + wrapper = _reader_wrapper(open_read_func) + builtins.open = wrapper + io.open = wrapper + _installed = True + + +def uninstall_open_patch() -> None: + global _installed + with _patch_lock: + if not _installed: + return + builtins.open = _ORIGINAL_OPEN + io.open = _ORIGINAL_IO_OPEN + _installed = False diff --git a/dlp_io/_raw.py b/dlp_io/_raw.py new file mode 100644 index 0000000..2574507 --- /dev/null +++ b/dlp_io/_raw.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +import io + +from ._errors import DlpTransportError +from ._helper import CHUNK_SIZE + + +class _PipeRawIO(io.RawIOBase): + _DISCARD_LIMIT = 64 * 1024 * 1024 + + def __init__(self, *, conn, size, path, offset, reopen, release) -> None: + super().__init__() + self._conn = conn + self._size = int(size) + self._path = path + self._pos = int(offset) + self._expected_end = max(self._size, self._pos) + self._reopen = reopen + self._release = release + self._buffer = bytearray() + self._eof = False + self._released = False + + def readable(self) -> bool: + return True + + def seekable(self) -> bool: + return True + + def tell(self) -> int: + if self.closed: + raise ValueError("I/O operation on closed file") + return self._pos + + def _receive(self) -> None: + try: + chunk = self._conn.recv_bytes(maxlength=CHUNK_SIZE) + except (EOFError, OSError) as exc: + raise DlpTransportError("helper disconnected before EOF") from exc + if chunk: + received_end = self._pos + len(self._buffer) + len(chunk) + if received_end > self._expected_end: + raise DlpTransportError( + "file size changed during transfer: expected end %d, received at least %d" + % (self._expected_end, received_end) + ) + self._buffer.extend(chunk) + return + if self._pos != self._expected_end: + raise DlpTransportError( + "file size changed during transfer: expected end %d, received %d" + % (self._expected_end, self._pos) + ) + self._eof = True + + def readinto(self, target) -> int: + if self.closed: + raise ValueError("I/O operation on closed file") + if len(target) == 0: + return 0 + if not self._buffer: + if self._eof: + return 0 + self._receive() + if self._eof: + return 0 + count = min(len(target), len(self._buffer)) + target[:count] = self._buffer[:count] + del self._buffer[:count] + self._pos += count + return count + + def seek(self, offset, whence=io.SEEK_SET) -> int: + if self.closed: + raise ValueError("I/O operation on closed file") + if whence == io.SEEK_SET: + target = offset + elif whence == io.SEEK_CUR: + target = self._pos + offset + elif whence == io.SEEK_END: + target = self._size + offset + else: + raise ValueError("invalid whence: %r" % (whence,)) + if target < 0: + raise ValueError("negative seek position: %d" % target) + delta = target - self._pos + if delta == 0: + return self._pos + if 0 < delta <= self._DISCARD_LIMIT: + remaining = delta + while remaining: + scratch = bytearray(min(remaining, CHUNK_SIZE)) + count = self.readinto(scratch) + if count == 0: + self._pos = target + self._expected_end = max(self._size, target) + break + remaining -= count + return self._pos + + self._conn.close() + try: + conn, size = self._reopen(self._path, target) + except Exception: + self.close() + raise + if int(size) != self._size: + conn.close() + self.close() + raise DlpTransportError( + "file size changed during seek: initial %d, reopened %d" + % (self._size, size) + ) + self._conn = conn + self._pos = target + self._expected_end = max(self._size, target) + self._buffer.clear() + self._eof = False + return self._pos + + def _release_once(self) -> None: + if self._released: + return + self._released = True + self._release(self) + + def close(self) -> None: + if not self.closed: + try: + self._conn.close() + finally: + self._release_once() + super().close() diff --git a/dlp_io/_session.py b/dlp_io/_session.py new file mode 100644 index 0000000..0b72818 --- /dev/null +++ b/dlp_io/_session.py @@ -0,0 +1,319 @@ +from __future__ import annotations + +import io +import json +import os +import queue +import shutil +import subprocess +import threading +from multiprocessing.context import AuthenticationError +from multiprocessing.connection import Client + +from ._errors import ( + DlpConfigurationError, + DlpHelperStartError, + DlpProtocolError, + DlpSessionBusyError, + DlpTransportError, +) +from ._helper import ( + CONTROL_LIMIT, + HANDSHAKE_PREFIX, + PROTOCOL_VERSION, + render_helper_source, +) +from ._raw import _PipeRawIO + +_SHUTDOWN_TIMEOUT = 5 + + +def _locate_python() -> list[str] | None: + for variable in ("DLP_IO_PYTHON", "DATAPACKER_DLP_PYTHON"): + value = os.environ.get(variable) + if value: + return [value] + executable = shutil.which("python") + if executable: + return [executable] + if shutil.which("py"): + return ["py", "-3"] + return None + + +def _helper_process_options(os_name=None) -> dict: + if (os_name or os.name) == "nt": + return {"creationflags": subprocess.CREATE_NO_WINDOW} + return {} + + +def _stop_process(process) -> None: + if process.poll() is not None: + return + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + pass + + +def _close_process_pipes(process) -> None: + for name in ("stdin", "stdout", "stderr"): + stream = getattr(process, name, None) + if stream is None: + continue + try: + stream.close() + except OSError: + pass + + +class DlpSession: + """One authenticated, short-lived approved-Python helper session.""" + + def __init__(self, process, pipe_name: str, authkey: bytes) -> None: + self._proc = process + self.pipe_name = pipe_name + self._authkey = authkey + self._request_lock = threading.Lock() + self._stream_lock = threading.Lock() + self._state_lock = threading.RLock() + self._active_raw = None + self._closed = False + + @classmethod + def start(cls, *, python_cmd=None, timeout=60): + if os.name != "nt": + raise DlpConfigurationError("dlp-io bridge reads require Windows AF_PIPE") + if timeout <= 0: + raise DlpConfigurationError("startup timeout must be positive") + command = list(python_cmd) if python_cmd else _locate_python() + if not command: + raise DlpHelperStartError( + "no approved Python found; configure DLP_IO_PYTHON" + ) + authkey = os.urandom(32) + source = render_helper_source(authkey) + try: + process = subprocess.Popen( + command + ["-u", "-"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + **_helper_process_options(), + ) + except OSError as exc: + raise DlpHelperStartError( + "failed to start approved Python %r: %s" % (command, exc) + ) from exc + try: + process.stdin.write(source.encode("utf-8")) + process.stdin.close() + except (BrokenPipeError, OSError) as exc: + _stop_process(process) + raise DlpHelperStartError("failed to inject helper source: %s" % exc) from exc + + result = queue.Queue() + + def read_handshake() -> None: + try: + result.put(process.stdout.readline()) + except Exception as exc: + result.put(exc) + + thread = threading.Thread(target=read_handshake, daemon=True) + thread.start() + try: + line = result.get(timeout=timeout) + except queue.Empty as exc: + _stop_process(process) + raise DlpHelperStartError( + "timed out waiting for helper handshake after %s seconds" % timeout + ) from exc + if isinstance(line, Exception): + _stop_process(process) + raise DlpHelperStartError("failed to read helper handshake: %s" % line) + decoded = line.decode("utf-8", "replace").strip() + if not decoded.startswith(HANDSHAKE_PREFIX): + _stop_process(process) + stderr = process.stderr.read().decode("utf-8", "replace").strip() + raise DlpHelperStartError( + "invalid helper handshake %r; stderr: %s" % (decoded, stderr) + ) + return cls(process, decoded[len(HANDSHAKE_PREFIX):], authkey) + + def _ensure_open(self) -> None: + if self._closed: + raise RuntimeError("dlp-io session is closed") + + def _validate_header(self, header) -> dict: + if not isinstance(header, dict): + raise DlpProtocolError("helper response must be a JSON object") + if header.get("version") != PROTOCOL_VERSION: + raise DlpProtocolError("helper response has an unsupported version") + if not isinstance(header.get("ok"), bool): + raise DlpProtocolError("helper response is missing boolean ok") + return header + + def _request(self, request): + with self._request_lock: + try: + encoded = json.dumps(request).encode("utf-8") + except (TypeError, ValueError) as exc: + raise DlpProtocolError("helper request is not JSON serializable") from exc + try: + conn = Client(self.pipe_name, family="AF_PIPE", authkey=self._authkey) + except AuthenticationError as exc: + raise DlpTransportError("failed to authenticate helper pipe") from exc + except (EOFError, OSError) as exc: + raise DlpTransportError("failed to connect to helper pipe") from exc + try: + try: + conn.send_bytes(encoded) + except (EOFError, OSError) as exc: + raise DlpTransportError("failed to send helper request") from exc + try: + payload = conn.recv_bytes(maxlength=CONTROL_LIMIT) + except (EOFError, OSError) as exc: + raise DlpTransportError("failed to receive helper response") from exc + try: + header = json.loads(payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise DlpProtocolError("helper response is not valid UTF-8 JSON") from exc + return conn, self._validate_header(header) + except Exception: + conn.close() + raise + + def ping(self) -> dict: + with self._state_lock: + self._ensure_open() + conn, header = self._request({"version": PROTOCOL_VERSION, "op": "ping"}) + conn.close() + if not header["ok"]: + raise DlpProtocolError(header.get("error", "ping failed")) + return header + + def _open_conn(self, path: str, offset: int = 0): + conn, header = self._request( + { + "version": PROTOCOL_VERSION, + "op": "read", + "path": path, + "offset": offset, + } + ) + if not header["ok"]: + conn.close() + raise OSError(header.get("error", "helper read failed")) + size = header.get("size") + if isinstance(size, bool) or not isinstance(size, int) or size < 0: + conn.close() + raise DlpProtocolError("helper returned an invalid file size") + return conn, size + + def _release_stream(self, raw) -> None: + with self._state_lock: + if self._active_raw is raw: + self._active_raw = None + self._stream_lock.release() + + def _reopen_stream(self, path: str, offset: int): + with self._state_lock: + self._ensure_open() + return self._open_conn(path, offset) + + def open_raw(self, path): + with self._state_lock: + self._ensure_open() + if not self._stream_lock.acquire(blocking=False): + raise DlpSessionBusyError("dlp-io session already has an active stream") + normalized = os.fsdecode(os.fspath(path)) + try: + conn, size = self._open_conn(normalized) + raw = _PipeRawIO( + conn=conn, + size=size, + path=normalized, + offset=0, + reopen=self._reopen_stream, + release=self._release_stream, + ) + with self._state_lock: + if self._closed: + raw.close() + raise RuntimeError("dlp-io session is closed") + self._active_raw = raw + return raw + except Exception: + self._stream_lock.release() + raise + + def open( + self, + file, + mode="r", + buffering=-1, + encoding=None, + errors=None, + newline=None, + ): + from ._api import _validate_path_mode, _wrap_reader + + buffering = _validate_path_mode(mode, buffering, encoding, errors, newline) + if "r" not in mode or "+" in mode: + raise io.UnsupportedOperation("DlpSession.open supports read-only modes") + + return _wrap_reader( + self.open_raw(file), mode, buffering, encoding, errors, newline + ) + + def open_read(self, path): + """Compatibility alias returning a buffered binary reader.""" + return self.open(path, "rb") + + def close(self) -> None: + with self._state_lock: + if self._closed: + return + self._closed = True + active = self._active_raw + if active is not None: + active.close() + + def request_quit() -> None: + try: + conn, header = self._request( + {"version": PROTOCOL_VERSION, "op": "quit"} + ) + conn.close() + if not header["ok"]: + raise DlpProtocolError(header.get("error", "helper quit failed")) + except Exception: + pass + + quit_thread = threading.Thread(target=request_quit, daemon=True) + quit_thread.start() + quit_thread.join(timeout=_SHUTDOWN_TIMEOUT) + try: + if quit_thread.is_alive(): + _stop_process(self._proc) + quit_thread.join(timeout=_SHUTDOWN_TIMEOUT) + else: + try: + self._proc.wait(timeout=_SHUTDOWN_TIMEOUT) + except subprocess.TimeoutExpired: + _stop_process(self._proc) + finally: + _close_process_pipes(self._proc) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + return False diff --git a/dlp_io/py.typed b/dlp_io/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/docs/DLP_IO_LIBRARY.md b/docs/DLP_IO_LIBRARY.md new file mode 100644 index 0000000..9521197 --- /dev/null +++ b/docs/DLP_IO_LIBRARY.md @@ -0,0 +1,145 @@ +# dlp-io + +`dlp-io` 为 Windows DLP 透明加密环境提供接近 `io.open` 的 Python 文件 API。非白名单进程负责业务逻辑和写入；文件读取由已获 DLP 白名单授权的 Python helper 完成，再通过经过认证的 Windows Named Pipe 返回明文字节。 + +当前版本为 `dlp-io==0.1.0`，distribution 名是 `dlp-io`，import 名是 `dlp_io`。 + +## 安装 + +当前 Gitea Python Package Registry 已开放匿名下载。Package 页面由 Gitea 根据上传后的 metadata 自动生成；本项目的 `.gitea/workflows/publish-dlp-io.yaml` 负责从 immutable tag 构建 wheel/sdist 并通过 `twine upload` 发布，不生成静态网页。 + +Package 页面： + +Windows PowerShell： + +```powershell +py -m pip install --index-url https://gitea.docker.antior.cn/api/packages/antior/pypi/simple dlp-io==0.1.0 +``` + +安装和下载不需要 token。上传新版本、覆盖管理和删除 Package 仍需 Gitea 凭据；发布凭据只能放在 workflow secret 中，不得写入命令行、URL 或 tracked 文件。 + +调用方需要安装 `dlp-io`。白名单 Python 无需安装该包：helper 源码由调用方以 UTF-8 通过 stdin 注入，且只依赖 Python 标准库，不会在磁盘创建临时 `.py` 文件。 + +运行环境必须是 Windows，并且机器上已有一个真正具备 DLP 明文读取权限的 Python。解释器查找顺序为： + +1. `DLP_IO_PYTHON` +2. `DATAPACKER_DLP_PYTHON`（兼容 DataPacker） +3. `python` +4. `py -3` + +生产环境优先把 `DLP_IO_PYTHON` 设置为审批过的 Python 绝对路径。 + +## io 风格用法 + +把原来的 `io.open` 或内置 `open` 改成 `dlp_io.open` 即可。纯读取走 helper；写入、追加和独占创建仍使用调用进程的原生文件 API。 + +```python +from pathlib import Path + +import dlp_io + + +source = Path(r"D:\Protected\input.txt") +output = Path(r"D:\Output\result.txt") + +with dlp_io.open(source, "r", encoding="utf-8", newline="") as reader: + text = reader.read() + +with dlp_io.open(output, "w", encoding="utf-8", newline="\n") as writer: + writer.write(text.upper()) + +dlp_io.shutdown() +``` + +`dlp_io.open()` 保留 `io.open()` 的参数名和位置习惯，包括 `buffering`、`encoding`、`errors`、`newline`、`closefd` 和 `opener`。 + +模式路由规则： + +- 路径型 `r`、`rt`、`rb`：通过 helper 读取。 +- 路径型 `w`、`a`、`x`：使用原生 `open` 写入。 +- 整数文件描述符或自定义 `opener`：使用原生 `open`。 +- 路径型 `+` 更新模式：显式抛出 `io.UnsupportedOperation`。 +- 路径读取配合 `closefd=False`：与标准 API 一样抛出 `ValueError`。 +- `rb` 配合 `buffering=0`：返回 raw stream；默认返回 buffered reader。 +- 文本模式的默认 encoding 与当前 Python `io.open` 一致；跨机器文件请显式传 `encoding="utf-8"`。 + +## 配置和生命周期 + +默认 session 在第一次读取时懒启动，并被后续读取复用： + +```python +import dlp_io + +dlp_io.configure( + python_executable=r"C:\ApprovedPython\python.exe", + startup_timeout=30, +) + +try: + with dlp_io.open(r"D:\Protected\payload.bin", "rb") as stream: + header = stream.read(64) +finally: + dlp_io.shutdown() +``` + +必须在默认 session 启动前调用 `configure()`。需要更换解释器时，先 `shutdown()`，再重新配置。`shutdown()` 和 session 的 `close()` 都是幂等操作。 + +## 显式 session + +库或服务代码可以显式持有 session，避免全局状态： + +```python +import dlp_io + +with dlp_io.DlpSession.start( + python_cmd=[r"C:\ApprovedPython\python.exe"], + timeout=30, +) as session: + with session.open(r"D:\Protected\archive.dat", "rb") as stream: + stream.seek(-32, 2) + trailer = stream.read() +``` + +一个 session 同时只允许一个活动读取 stream。请关闭当前 stream 后再打开下一个；并行读取应由调用方创建彼此独立的 session。 + +## 兼容既有代码的全局 patch + +只有当第三方代码无法逐个改为 `dlp_io.open()` 时，才临时 patch `builtins.open` 和 `io.open`： + +```python +import dlp_io + +dlp_io.configure(python_executable=r"C:\ApprovedPython\python.exe") +dlp_io.install_open_patch() +try: + run_legacy_application() +finally: + dlp_io.uninstall_open_patch() + dlp_io.shutdown() +``` + +在打包入口中，应先 import 完业务模块和本地资源，再安装 patch。patch 只路由路径型纯读模式，重复安装和卸载是幂等的。它不会拦截 `os.open`，也无法拦截 C 扩展内部绕过 Python `open` 的读取。 + +如果应用自己管理 session，可以把 `session.open_raw` 传给 `install_reader_patch()`。 + +## 错误和安全边界 + +主要异常均继承 `DlpIoError`： + +- `DlpConfigurationError`：平台、配置或参数不成立。 +- `DlpHelperStartError`：白名单 Python 启动、源码注入或握手失败。 +- `DlpProtocolError`：协议版本、控制消息或响应结构无效。 +- `DlpTransportError`：helper 在显式 EOF 前断开，或文件传输被截断。 +- `DlpSessionBusyError`：同一个 session 已有活动 stream。 + +helper 启动或读取失败时，库会显式报错并且不回退到当前 EXE 直接读取。DLP 直读可能返回合法长度的密文；静默回退会把数据损坏伪装成成功。 + +版本 1 会在 EOF 和 seek 重连时检测文件 size 的截断或增长，但不提供文件快照，也不能识别读取期间发生的同尺寸内容替换；调用方应避免并发修改输入文件。 + +每次 session 使用随机 32-byte authkey，AF_PIPE 连接执行 challenge-response 认证；协议控制消息和数据帧都有明确大小边界，零长度数据帧是唯一 EOF。 + +## 卸载 + +```powershell +py -m pip uninstall dlp-io +``` diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..3106dbf --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,33 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +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.9" +classifiers = [ + "Development Status :: 3 - Alpha", + "Operating System :: Microsoft :: Windows", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", +] + +[project.urls] +Repository = "https://gitea.docker.antior.cn/antior/dlp-io" + +[tool.setuptools] +packages = ["dlp_io"] +include-package-data = true + +[tool.setuptools.dynamic] +version = {attr = "dlp_io.__version__"} + +[tool.setuptools.package-data] +dlp_io = ["py.typed"] diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..da636be --- /dev/null +++ b/pytest.ini @@ -0,0 +1,4 @@ +[pytest] +testpaths = tests +python_files = test_*.py +addopts = -ra diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..1978f20 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,7 @@ +build +packaging +pytest +pytest-cov +setuptools>=68,<82 +tomli; python_version < "3.11" +wheel diff --git a/tests/test_dlp_io_api.py b/tests/test_dlp_io_api.py new file mode 100644 index 0000000..1d5b387 --- /dev/null +++ b/tests/test_dlp_io_api.py @@ -0,0 +1,280 @@ +from __future__ import annotations + +import inspect +import io +import builtins +from pathlib import Path + +import pytest + +import dlp_io +import dlp_io._api as api_module + + +class _FakeSession: + def __init__(self, payload: bytes = b"bridged data") -> None: + self.payload = payload + self.paths: list[str] = [] + self.closed = False + + def open_raw(self, path): + self.paths.append(str(path)) + return io.BytesIO(self.payload) + + def close(self) -> None: + self.closed = True + + +@pytest.fixture(autouse=True) +def reset_default_session(): + dlp_io.shutdown() + yield + dlp_io.shutdown() + + +def test_open_signature_matches_io_open() -> None: + assert inspect.signature(dlp_io.open) == inspect.signature(io.open) + + +def test_write_mode_uses_native_open(tmp_path: Path) -> None: + output = tmp_path / "output.txt" + + with dlp_io.open(output, "w", encoding="utf-8") as file: + file.write("本地写入") + + assert output.read_bytes() == "本地写入".encode("utf-8") + + +def test_path_update_mode_is_explicitly_unsupported(tmp_path: Path) -> None: + path = tmp_path / "data.bin" + path.write_bytes(b"data") + + with pytest.raises(io.UnsupportedOperation): + dlp_io.open(path, "r+b") + + +def test_path_rejects_closefd_false(tmp_path: Path) -> None: + path = tmp_path / "data.bin" + + with pytest.raises(ValueError): + dlp_io.open(path, "rb", closefd=False) + + +def test_custom_opener_uses_native_open(tmp_path: Path) -> None: + path = tmp_path / "native.txt" + path.write_text("native", encoding="utf-8") + calls = [] + + with api_module._ORIGINAL_OPEN(path, "rb") as expected: + expected_bytes = expected.read() + + # Keep the returned fd alive independently; os.open is the normal opener contract. + import os + + def os_opener(name, flags): + calls.append((name, flags)) + return os.open(name, flags) + + with dlp_io.open(path, "rb", opener=os_opener) as file: + assert file.read() == expected_bytes + + assert len(calls) == 1 + + +def test_custom_opener_update_mode_uses_native_open(tmp_path: Path) -> None: + import os + + path = tmp_path / "native-update.bin" + path.write_bytes(b"abc") + + with dlp_io.open(path, "r+b", opener=os.open) as file: + assert file.read(1) == b"a" + file.seek(0) + file.write(b"z") + + assert path.read_bytes() == b"zbc" + + +def test_integer_fd_uses_native_open(tmp_path: Path) -> None: + import os + + path = tmp_path / "fd.bin" + path.write_bytes(b"fd data") + descriptor = os.open(path, os.O_RDONLY) + + with dlp_io.open(descriptor, "rb", closefd=True) as file: + assert file.read() == b"fd data" + + +def test_read_lazily_starts_and_reuses_default_session(monkeypatch) -> None: + created: list[tuple[object, float, _FakeSession]] = [] + + def fake_start(*, python_cmd=None, timeout=60): + session = _FakeSession() + created.append((python_cmd, timeout, session)) + return session + + monkeypatch.setattr(api_module.DlpSession, "start", fake_start) + dlp_io.configure(python_executable=r"C:\Approved\python.exe", startup_timeout=12) + + with dlp_io.open(r"D:\data\first.bin", "rb") as first: + assert first.read() == b"bridged data" + with dlp_io.open(r"D:\data\second.bin", "rb") as second: + assert second.read() == b"bridged data" + + assert len(created) == 1 + assert tuple(Path(item) for item in created[0][0]) == ( + Path(r"C:\Approved\python.exe"), + ) + assert created[0][1] == 12 + assert tuple(Path(item) for item in created[0][2].paths) == ( + Path(r"D:\data\first.bin"), + Path(r"D:\data\second.bin"), + ) + + +def test_configure_rejects_active_session_and_allows_after_shutdown(monkeypatch) -> None: + session = _FakeSession() + monkeypatch.setattr(api_module.DlpSession, "start", lambda **kwargs: session) + + with dlp_io.open(r"D:\data\input.bin", "rb") as file: + assert file.read() == b"bridged data" + + with pytest.raises(dlp_io.DlpConfigurationError): + dlp_io.configure(python_executable=r"C:\Other\python.exe") + + dlp_io.shutdown() + assert session.closed + + dlp_io.configure(python_executable=r"C:\Other\python.exe") + + +def test_binary_buffering_zero_returns_raw_stream(monkeypatch) -> None: + session = _FakeSession(b"raw") + monkeypatch.setattr(api_module.DlpSession, "start", lambda **kwargs: session) + + with dlp_io.open(r"D:\data\raw.bin", "rb", buffering=0) as file: + assert isinstance(file, io.BytesIO) + assert file.read() == b"raw" + + +def test_binary_default_returns_buffered_reader(monkeypatch) -> None: + session = _FakeSession(b"buffered") + monkeypatch.setattr(api_module.DlpSession, "start", lambda **kwargs: session) + + with dlp_io.open(r"D:\data\buffered.bin", "rb") as file: + assert isinstance(file, io.BufferedReader) + assert file.read() == b"buffered" + + +def test_text_mode_preserves_encoding_errors_and_newline(monkeypatch) -> None: + session = _FakeSession("第一行\r\n第二行".encode("utf-8")) + monkeypatch.setattr(api_module.DlpSession, "start", lambda **kwargs: session) + + with dlp_io.open( + r"D:\data\text.txt", + "r", + encoding="utf-8", + errors="strict", + newline="", + ) as file: + assert file.read().encode("utf-8") == session.payload + + +@pytest.mark.parametrize( + "kwargs", + [ + {"encoding": "utf-8"}, + {"errors": "strict"}, + {"newline": ""}, + ], +) +def test_binary_mode_rejects_text_arguments_before_starting_helper( + monkeypatch, kwargs +) -> None: + monkeypatch.setattr( + api_module.DlpSession, + "start", + lambda **unused: pytest.fail("invalid arguments must not start the helper"), + ) + + with pytest.raises(ValueError): + dlp_io.open(r"D:\data\payload.bin", "rb", **kwargs) + + +@pytest.mark.parametrize("mode", ["", "rr", "rbt", "U", "q"]) +def test_invalid_read_mode_is_rejected_before_starting_helper( + monkeypatch, mode: str +) -> None: + monkeypatch.setattr( + api_module.DlpSession, + "start", + lambda **unused: pytest.fail("invalid mode must not start the helper"), + ) + + with pytest.raises(ValueError): + dlp_io.open(r"D:\data\payload.bin", mode) + + +@pytest.mark.parametrize("buffering", [None, "invalid"]) +def test_invalid_buffering_is_rejected_before_starting_helper( + monkeypatch, buffering +) -> None: + monkeypatch.setattr( + api_module.DlpSession, + "start", + lambda **unused: pytest.fail("invalid buffering must not start the helper"), + ) + + with pytest.raises(TypeError): + dlp_io.open(r"D:\data\payload.bin", "rb", buffering=buffering) + + +def test_negative_buffering_is_rejected_before_starting_helper(monkeypatch) -> None: + monkeypatch.setattr( + api_module.DlpSession, + "start", + lambda **unused: pytest.fail("invalid buffering must not start the helper"), + ) + + with pytest.raises(ValueError): + dlp_io.open(r"D:\data\payload.bin", "rb", buffering=-2) + + +def test_invalid_pathlike_is_rejected_before_starting_helper(monkeypatch) -> None: + class InvalidPath: + def __fspath__(self): + raise TypeError("invalid path") + + monkeypatch.setattr( + api_module.DlpSession, + "start", + lambda **unused: pytest.fail("invalid path must not start the helper"), + ) + + with pytest.raises(TypeError): + dlp_io.open(InvalidPath(), "rb") + + +def test_open_patch_routes_reads_and_keeps_native_writes(tmp_path, monkeypatch) -> None: + session = _FakeSession(b"patched") + monkeypatch.setattr(api_module.DlpSession, "start", lambda **kwargs: session) + original_builtin = builtins.open + original_io = io.open + + dlp_io.install_open_patch() + dlp_io.install_open_patch() + try: + with builtins.open(r"D:\data\patched.bin", "rb") as file: + assert file.read() == b"patched" + output = tmp_path / "native.txt" + with builtins.open(output, "w", encoding="utf-8") as file: + file.write("native") + with original_builtin(output, "rb") as file: + assert file.read() == b"native" + finally: + dlp_io.uninstall_open_patch() + dlp_io.uninstall_open_patch() + + assert builtins.open is original_builtin + assert io.open is original_io diff --git a/tests/test_dlp_io_packaging.py b/tests/test_dlp_io_packaging.py new file mode 100644 index 0000000..0a56244 --- /dev/null +++ b/tests/test_dlp_io_packaging.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +import importlib +import shutil +import subprocess +import sys +import tarfile +import zipfile +from dataclasses import dataclass +from enum import Enum +from pathlib import Path, PurePosixPath + +import pytest +from packaging.specifiers import SpecifierSet +from packaging.utils import canonicalize_name, parse_wheel_filename +from packaging.version import Version + +import dlp_io + +try: + import tomllib +except ModuleNotFoundError: # Python 3.9-3.10 + import tomli as tomllib + + +PROJECT_ROOT = Path(__file__).resolve().parents[1] + + +class DynamicField(Enum): + VERSION = "version" + + +class PackageName(Enum): + DLP_IO = canonicalize_name("dlp-io") + + +class PackageClassifier(Enum): + WINDOWS = "Operating System :: Microsoft :: Windows" + + +@dataclass(frozen=True) +class BuiltDistributions: + wheel: Path + sdist: Path + + +def _load_pyproject() -> dict: + with (PROJECT_ROOT / "pyproject.toml").open("rb") as file: + return tomllib.load(file) + + +@pytest.fixture(scope="module") +def built_distributions(tmp_path_factory: pytest.TempPathFactory) -> BuiltDistributions: + root = tmp_path_factory.mktemp("dlp-io-build") + source = root / "source" + source.mkdir() + shutil.copy2(PROJECT_ROOT / "pyproject.toml", source) + shutil.copy2(PROJECT_ROOT / "MANIFEST.in", source) + shutil.copytree(PROJECT_ROOT / "dlp_io", source / "dlp_io") + shutil.copytree(PROJECT_ROOT / "docs", source / "docs") + + dist = root / "dist" + completed = subprocess.run( + [ + sys.executable, + "-m", + "build", + "--no-isolation", + "--outdir", + str(dist), + ], + cwd=source, + check=False, + capture_output=True, + ) + assert completed.returncode == 0 + + wheels = tuple(dist.glob("*.whl")) + sdists = tuple(dist.glob("*.tar.gz")) + assert len(wheels) == 1 + assert len(sdists) == 1 + return BuiltDistributions(wheel=wheels[0], sdist=sdists[0]) + + +def test_pyproject_declares_structured_public_package_contract() -> None: + config = _load_pyproject() + project = config["project"] + setuptools = config["tool"]["setuptools"] + + assert PackageName(canonicalize_name(project["name"])) is PackageName.DLP_IO + assert SpecifierSet(project["requires-python"]) == SpecifierSet(">=3.9") + assert {DynamicField(item) for item in project["dynamic"]} == { + DynamicField.VERSION + } + known_classifier_values = {item.value for item in PackageClassifier} + known_classifiers = { + PackageClassifier(item) + for item in project["classifiers"] + if item in known_classifier_values + } + assert PackageClassifier.WINDOWS in known_classifiers + assert {Path(item) for item in setuptools["packages"]} == {Path("dlp_io")} + assert {Path(item) for item in setuptools["package-data"]["dlp_io"]} == { + Path("py.typed") + } + + version_ref = config["tool"]["setuptools"]["dynamic"]["version"]["attr"] + module_name, attribute_name = version_ref.rsplit(".", 1) + declared_version = getattr(importlib.import_module(module_name), attribute_name) + assert Version(declared_version) == Version(dlp_io.__version__) + + +def test_wheel_contains_only_installable_dlp_io_runtime( + built_distributions: BuiltDistributions, +) -> None: + name, version, build, tags = parse_wheel_filename(built_distributions.wheel.name) + with zipfile.ZipFile(built_distributions.wheel) as archive: + members = {PurePosixPath(item) for item in archive.namelist()} + + assert PackageName(name) is PackageName.DLP_IO + assert version == Version(dlp_io.__version__) + assert build == () + assert tags + assert PurePosixPath("dlp_io/__init__.py") in members + assert PurePosixPath("dlp_io/py.typed") in members + member_roots = {Path(item.parts[0]) for item in members} + assert member_roots.isdisjoint({Path("tests"), Path("Lib")}) + assert PurePosixPath(".env") not in members + + +def test_sdist_contains_package_sources_and_public_readme_only( + built_distributions: BuiltDistributions, +) -> None: + with tarfile.open(built_distributions.sdist, mode="r:gz") as archive: + members = { + PurePosixPath(*PurePosixPath(item.name).parts[1:]) + for item in archive.getmembers() + if item.isfile() + } + + assert PurePosixPath("pyproject.toml") in members + assert PurePosixPath("docs/DLP_IO_LIBRARY.md") in members + assert PurePosixPath("dlp_io/__init__.py") in members + member_roots = {Path(item.parts[0]) for item in members} + assert Path("tests") not in member_roots + assert PurePosixPath("README.md") not in members + assert PurePosixPath(".env") not in members diff --git a/tests/test_dlp_io_session.py b/tests/test_dlp_io_session.py new file mode 100644 index 0000000..a942664 --- /dev/null +++ b/tests/test_dlp_io_session.py @@ -0,0 +1,408 @@ +from __future__ import annotations + +import io +import os +import subprocess +import sys +import threading +import time +from multiprocessing.context import AuthenticationError +from multiprocessing.connection import Client + +import pytest + +from dlp_io import ( + DlpConfigurationError, + DlpProtocolError, + DlpSession, + DlpSessionBusyError, + DlpTransportError, +) +import dlp_io._session as session_module +from dlp_io._helper import ( + CHUNK_SIZE, + HelperErrorCode, + HelperSourceConfig, + render_helper_source, +) +from dlp_io._raw import _PipeRawIO + + +pytestmark = pytest.mark.skipif(os.name != "nt", reason="AF_PIPE requires Windows") + + +@pytest.fixture +def session(): + current = DlpSession.start(python_cmd=[sys.executable], timeout=15) + yield current + current.close() + + +def test_helper_round_trip_uses_multiple_chunks(tmp_path, session) -> None: + payload = os.urandom(3 * 1024 * 1024) + path = tmp_path / "payload.bin" + path.write_bytes(payload) + + with session.open(path, "rb") as file: + assert file.read() == payload + + +def test_helper_missing_file_is_explicit(tmp_path, session) -> None: + with pytest.raises(OSError): + session.open_raw(tmp_path / "missing.bin") + + +def test_helper_stream_is_seekable(tmp_path, session) -> None: + payload = bytes(range(256)) * 2048 + path = tmp_path / "seek.bin" + path.write_bytes(payload) + + with session.open(path, "rb") as file: + assert file.read(10) == payload[:10] + assert file.seek(100) == 100 + assert file.read(10) == payload[100:110] + assert file.seek(5) == 5 + assert file.read(10) == payload[5:15] + assert file.seek(-7, io.SEEK_END) == len(payload) - 7 + assert file.read() == payload[-7:] + + +def test_session_rejects_second_active_stream(tmp_path, session) -> None: + path = tmp_path / "busy.bin" + path.write_bytes(b"busy") + + first = session.open_raw(path) + try: + with pytest.raises(DlpSessionBusyError): + session.open_raw(path) + finally: + first.close() + + with session.open(path, "rb") as reopened: + assert reopened.read() == b"busy" + + +def test_open_raw_rechecks_closed_state_after_connection(monkeypatch) -> None: + process = _HungProcess() + current = DlpSession(process, r"\\.\pipe\race", b"a" * 32) + connection = _ClosableConnection() + entered = threading.Event() + proceed = threading.Event() + outcome = [] + + def delayed_open(path): + entered.set() + proceed.wait(timeout=5) + return connection, 1 + + monkeypatch.setattr(current, "_open_conn", delayed_open) + + def run_open() -> None: + try: + outcome.append(current.open_raw("race.bin")) + except Exception as exc: + outcome.append(exc) + + thread = threading.Thread(target=run_open) + thread.start() + assert entered.wait(timeout=1) + with current._state_lock: + current._closed = True + proceed.set() + thread.join(timeout=1) + + assert len(outcome) == 1 + assert isinstance(outcome[0], RuntimeError) + assert connection.closed + assert current._stream_lock.acquire(blocking=False) + current._stream_lock.release() + + +def test_wrong_authkey_is_rejected_without_killing_helper(session) -> None: + with pytest.raises(AuthenticationError): + Client(session.pipe_name, family="AF_PIPE", authkey=b"wrong" * 8) + + assert session.ping()["version"] == 1 + + +def test_protocol_rejects_wrong_version_and_negative_offset(session) -> None: + conn, header = session._request({"version": 99, "op": "ping"}) + conn.close() + assert header["ok"] is False + assert HelperErrorCode(header["error_code"]) is HelperErrorCode.UNSUPPORTED_VERSION + + conn, header = session._request( + {"version": 1, "op": "read", "path": "data.bin", "offset": -1} + ) + conn.close() + assert header["ok"] is False + assert HelperErrorCode(header["error_code"]) is HelperErrorCode.INVALID_OFFSET + + +def test_helper_source_requires_32_byte_authkey() -> None: + with pytest.raises(DlpConfigurationError): + render_helper_source(b"short") + + authkey = bytes(range(32)) + config = HelperSourceConfig.from_authkey(authkey) + + assert config.authkey == authkey + assert config.protocol_version == 1 + assert config.control_limit == 64 * 1024 + assert config.chunk_size == 1024 * 1024 + + +class _BrokenConnection: + def recv_bytes(self, maxlength=None): + raise EOFError("pipe closed") + + def close(self): + return None + + +class _TruncatedConnection: + def __init__(self) -> None: + self._chunks = iter((b"abc", b"")) + + def recv_bytes(self, maxlength=None): + return next(self._chunks) + + def close(self): + return None + + +class _GrowingConnection: + def recv_bytes(self, maxlength=None): + return b"ab" + + def close(self): + return None + + +def test_unexpected_disconnect_is_not_eof() -> None: + raw = _PipeRawIO( + conn=_BrokenConnection(), + size=3, + path="broken.bin", + offset=0, + reopen=lambda path, offset: None, + release=lambda raw: None, + ) + + with pytest.raises(DlpTransportError): + raw.readinto(bytearray(3)) + + +def test_zero_length_readinto_does_not_touch_connection() -> None: + raw = _PipeRawIO( + conn=_BrokenConnection(), + size=3, + path="zero.bin", + offset=0, + reopen=lambda path, offset: None, + release=lambda raw: None, + ) + + assert raw.readinto(bytearray()) == 0 + + +def test_explicit_eof_detects_truncated_file() -> None: + raw = _PipeRawIO( + conn=_TruncatedConnection(), + size=4, + path="truncated.bin", + offset=0, + reopen=lambda path, offset: None, + release=lambda raw: None, + ) + + assert raw.readinto(bytearray(3)) == 3 + with pytest.raises(DlpTransportError): + raw.readinto(bytearray(1)) + + +def test_frame_cannot_exceed_helper_reported_file_size() -> None: + raw = _PipeRawIO( + conn=_GrowingConnection(), + size=1, + path="growing.bin", + offset=0, + reopen=lambda path, offset: None, + release=lambda raw: None, + ) + + with pytest.raises(DlpTransportError): + raw.readinto(bytearray(1)) + + +class _OversizedConnection: + def __init__(self) -> None: + self.maxlength = None + + def recv_bytes(self, maxlength=None): + self.maxlength = maxlength + raise OSError("bad message length") + + def close(self): + return None + + +def test_data_frame_is_bounded_and_oversize_is_transport_error() -> None: + connection = _OversizedConnection() + raw = _PipeRawIO( + conn=connection, + size=1, + path="oversized.bin", + offset=0, + reopen=lambda path, offset: None, + release=lambda raw: None, + ) + + with pytest.raises(DlpTransportError): + raw.readinto(bytearray(1)) + + assert connection.maxlength == CHUNK_SIZE + + +class _ClosableConnection: + def __init__(self) -> None: + self.closed = False + + def close(self) -> None: + self.closed = True + + +def test_seek_reopen_rejects_file_size_change() -> None: + initial = _ClosableConnection() + reopened = _ClosableConnection() + released = [] + raw = _PipeRawIO( + conn=initial, + size=100, + path="changing.bin", + offset=10, + reopen=lambda path, offset: (reopened, 50), + release=released.append, + ) + + with pytest.raises(DlpTransportError): + raw.seek(0) + + assert initial.closed + assert reopened.closed + assert raw.closed + assert released == [raw] + + +class _ControlConnection: + def __init__(self, response=None, error=None) -> None: + self.response = response + self.error = error + self.closed = False + + def send_bytes(self, payload) -> None: + return None + + def recv_bytes(self, maxlength=None): + if self.error is not None: + raise self.error + return self.response + + def close(self) -> None: + self.closed = True + + +def _unstarted_session() -> DlpSession: + return DlpSession(process=object(), pipe_name=r"\\.\pipe\test", authkey=b"a" * 32) + + +def test_request_maps_authentication_failure_to_transport_error(monkeypatch) -> None: + def reject(*args, **kwargs): + raise AuthenticationError("digest rejected") + + monkeypatch.setattr(session_module, "Client", reject) + + with pytest.raises(DlpTransportError): + _unstarted_session()._request({"version": 1, "op": "ping"}) + + +def test_request_maps_disconnect_and_invalid_json_to_public_errors(monkeypatch) -> None: + disconnected = _ControlConnection(error=EOFError("closed")) + monkeypatch.setattr(session_module, "Client", lambda *args, **kwargs: disconnected) + with pytest.raises(DlpTransportError): + _unstarted_session()._request({"version": 1, "op": "ping"}) + assert disconnected.closed + + invalid = _ControlConnection(response=b"not-json") + monkeypatch.setattr(session_module, "Client", lambda *args, **kwargs: invalid) + with pytest.raises(DlpProtocolError): + _unstarted_session()._request({"version": 1, "op": "ping"}) + assert invalid.closed + + +def test_session_close_is_idempotent() -> None: + current = DlpSession.start(python_cmd=[sys.executable], timeout=15) + process = current._proc + + current.close() + current.close() + + assert process.poll() is not None + assert process.stdin.closed + assert process.stdout.closed + assert process.stderr.closed + + +class _HungProcess: + def __init__(self) -> None: + self.returncode = None + self.terminated = False + self.killed = False + self.stdin = io.BytesIO() + self.stdout = io.BytesIO() + self.stderr = io.BytesIO() + + def poll(self): + return self.returncode + + def terminate(self) -> None: + self.terminated = True + self.returncode = 1 + + def kill(self) -> None: + self.killed = True + self.returncode = 1 + + def wait(self, timeout=None): + if self.returncode is None: + raise subprocess.TimeoutExpired("helper", timeout) + return self.returncode + + +def test_session_close_bounds_hung_quit_and_closes_process_pipes(monkeypatch) -> None: + process = _HungProcess() + current = DlpSession(process, r"\\.\pipe\hung", b"a" * 32) + release = threading.Event() + entered = threading.Event() + + def hung_request(request): + entered.set() + release.wait(timeout=10) + raise OSError("released") + + monkeypatch.setattr(current, "_request", hung_request) + monkeypatch.setattr(session_module, "_SHUTDOWN_TIMEOUT", 0.01) + + started = time.monotonic() + try: + current.close() + finally: + release.set() + elapsed = time.monotonic() - started + + assert entered.is_set() + assert elapsed < 1 + assert process.terminated + assert process.stdin.closed + assert process.stdout.closed + assert process.stderr.closed