From fe1a6ed74992d92a0c146b1bf6cbf9ea530cd1da Mon Sep 17 00:00:00 2001 From: antior Date: Fri, 31 Jul 2026 16:42:59 +0800 Subject: [PATCH] Consolidate to single dlp_io module and publish release artifacts via Gitea Actions --- .gitea/workflows/publish-dlp-io.yaml | 450 +++++-------- MANIFEST.in | 1 - README.md | 18 +- dlp_io.py | 903 +++++++++++++++++++++++++++ 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 | 21 +- pyproject.toml | 11 +- requirements-dev.txt | 2 + setup.py | 33 + tests/test_dlp_io_api.py | 2 +- tests/test_dlp_io_packaging.py | 59 +- tests/test_dlp_io_session.py | 6 +- 20 files changed, 1191 insertions(+), 1245 deletions(-) create mode 100644 dlp_io.py delete mode 100644 dlp_io/__init__.py delete mode 100644 dlp_io/_api.py delete mode 100644 dlp_io/_config.py delete mode 100644 dlp_io/_errors.py delete mode 100644 dlp_io/_helper.py delete mode 100644 dlp_io/_patch.py delete mode 100644 dlp_io/_raw.py delete mode 100644 dlp_io/_session.py delete mode 100644 dlp_io/py.typed create mode 100644 setup.py diff --git a/.gitea/workflows/publish-dlp-io.yaml b/.gitea/workflows/publish-dlp-io.yaml index a905776..44329bd 100644 --- a/.gitea/workflows/publish-dlp-io.yaml +++ b/.gitea/workflows/publish-dlp-io.yaml @@ -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.9-3.13 + - name: Test Python 3.10-3.14 shell: pwsh run: | $ErrorActionPreference = "Stop" - foreach ($pythonVersion in @("3.9", "3.10", "3.11", "3.12", "3.13")) { + foreach ($pythonVersion in @("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 @@ -118,20 +118,57 @@ jobs: 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" ) + foreach ($pythonVersion in @("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" + try { + & py "-$pythonVersion" -m build --wheel --no-isolation + if ($LASTEXITCODE -ne 0) { throw "pyd build failed on $pythonVersion" } + } finally { + Remove-Item Env:DLP_IO_PYD + } + if (-not (Test-Path -LiteralPath (Join-Path dist $wheel) -PathType Leaf)) { + throw "expected pyd wheel missing: $wheel" + } + $expected += $wheel + } + + py -3.13 -m twine check "dist\dlp_io-$($env:DLP_IO_PACKAGE_VERSION)*" + if ($LASTEXITCODE -ne 0) { throw "twine check failed" } $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 + $assetsDir = (New-Item -ItemType Directory -Force assets).FullName + Copy-Item -LiteralPath (Resolve-Path dlp_io.py) -Destination $assetsDir + Add-Type -AssemblyName System.IO.Compression.FileSystem + foreach ($wheelFile in @(Get-ChildItem -LiteralPath dist -Filter "*-cp3*-win_amd64.whl" -File)) { + $zip = [System.IO.Compression.ZipFile]::OpenRead($wheelFile.FullName) + try { + $pydEntries = @($zip.Entries | Where-Object { $_.FullName -like "*.pyd" }) + if ($pydEntries.Count -ne 1) { + throw "pyd wheel must contain exactly one .pyd: $($wheelFile.Name)" + } + $target = Join-Path $assetsDir $pydEntries[0].Name + [System.IO.Compression.ZipFileExtensions]::ExtractToFile($pydEntries[0], $target, $true) + } finally { + $zip.Dispose() + } + } + $bareAssets = @(Get-ChildItem -LiteralPath $assetsDir -File | ForEach-Object Name) + if ($bareAssets.Count -ne 6) { + throw "bare asset set mismatch: $($bareAssets -join ', ')" + } + + - name: Smoke test release artifacts shell: pwsh run: | $ErrorActionPreference = "Stop" @@ -145,8 +182,8 @@ jobs: 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 + $pureWheel = (Resolve-Path (Join-Path dist "dlp_io-$($env:DLP_IO_PACKAGE_VERSION)-py3-none-any.whl")).Path + & $python -m pip install --no-index $pureWheel --progress-bar off if ($LASTEXITCODE -ne 0) { throw "wheel installation failed" } Push-Location $rootFull try { @@ -161,293 +198,146 @@ jobs: } } - - 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) { + foreach ($pythonVersion in @("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) + $pydRootFull = [IO.Path]::GetFullPath($pydRoot) + if (-not $pydRootFull.StartsWith($tempFull, [StringComparison]::OrdinalIgnoreCase)) { + throw "unsafe pyd smoke path: $pydRootFull" + } 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" + & py "-$pythonVersion" -m venv $pydRootFull + if ($LASTEXITCODE -ne 0) { throw "pyd smoke venv creation failed on $pythonVersion" } + $pydPython = Join-Path $pydRootFull "Scripts\python.exe" + & $pydPython -m pip install --no-index $pydWheel --progress-bar off + if ($LASTEXITCODE -ne 0) { throw "pyd wheel installation failed on $pythonVersion" } + Push-Location $pydRootFull + try { + & $pydPython -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('pyd smoke passed on $pythonVersion')" + if ($LASTEXITCODE -ne 0) { throw "pyd wheel smoke failed on $pythonVersion" } + } finally { + Pop-Location } - $builder = [UriBuilder]$uri - $builder.Fragment = "" - [pscustomobject]@{ - filename = [Net.WebUtility]::HtmlDecode($match.Groups["name"].Value) - url = $builder.Uri.AbsoluteUri - sha256 = $hashMatch.Groups["hash"].Value.ToLowerInvariant() + } finally { + if (Test-Path -LiteralPath $pydRootFull) { + Remove-Item -LiteralPath $pydRootFull -Recurse -Force } } } - 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" + $bareRoot = Join-Path $env:TEMP ("dlp-io-bare-smoke-" + $PID) + $bareRootFull = [IO.Path]::GetFullPath($bareRoot) + if (-not $bareRootFull.StartsWith($tempFull, [StringComparison]::OrdinalIgnoreCase)) { + throw "unsafe bare smoke path: $bareRootFull" } 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 + foreach ($pythonVersion in @("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 + Copy-Item -LiteralPath (Resolve-Path (Join-Path assets "dlp_io.cp$abi-win_amd64.pyd")) -Destination $sandbox + Push-Location $sandbox + try { + & py "-$pythonVersion" -c "import dlp_io,sys,tempfile; from pathlib import Path; assert dlp_io.__version__ == '$env:DLP_IO_PACKAGE_VERSION'; assert dlp_io.__file__.endswith('.pyd'); 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('bare pyd smoke passed on $pythonVersion')" + if ($LASTEXITCODE -ne 0) { throw "bare pyd smoke failed on $pythonVersion" } + } finally { + Pop-Location } - 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) { + $pySandbox = Join-Path $bareRootFull "py" + New-Item -ItemType Directory -Force $pySandbox | Out-Null + Copy-Item -LiteralPath (Resolve-Path (Join-Path assets "dlp_io.py")) -Destination $pySandbox + Push-Location $pySandbox 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" } + & py -3.13 -c "import dlp_io,sys,tempfile; from pathlib import Path; assert dlp_io.__version__ == '$env:DLP_IO_PACKAGE_VERSION'; assert dlp_io.__file__.endswith('.py'); 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('bare py smoke passed')" + if ($LASTEXITCODE -ne 0) { throw "bare py smoke failed" } } finally { Pop-Location } } finally { - if (Test-Path -LiteralPath $registryRootFull) { - Remove-Item -LiteralPath $registryRootFull -Recurse -Force + if (Test-Path -LiteralPath $bareRootFull) { + Remove-Item -LiteralPath $bareRootFull -Recurse -Force } } - Write-Host "registry verified: dlp-io==$packageVersion ($($actual.Count) files)" + + - name: Create Gitea release with wheels + shell: pwsh + env: + GITEA_TOKEN: ${{ secrets.GITHUB_TOKEN }} + GITEA_SERVER: ${{ gitea.server_url }} + GITEA_REPOSITORY: ${{ github.repository }} + run: | + $ErrorActionPreference = "Stop" + if ([string]::IsNullOrWhiteSpace($env:GITEA_TOKEN)) { + throw "GITHUB_TOKEN is required to publish the release" + } + $tag = $env:DLP_IO_PACKAGE_TAG + $version = $env:DLP_IO_PACKAGE_VERSION + $api = "$env:GITEA_SERVER/api/v1/repos/$env:GITEA_REPOSITORY" + $headers = @{ Authorization = "token $env:GITEA_TOKEN" } + + $artifacts = @(Get-ChildItem -LiteralPath dist -File) + @(Get-ChildItem -LiteralPath assets -File) + if ($artifacts.Count -eq 0) { throw "no distribution files to publish" } + + $release = $null + try { + $release = Invoke-RestMethod -Headers $headers -Uri "$api/releases/tags/$tag" + } catch { + if ([int]$_.Exception.Response.StatusCode -ne 404) { throw } + } + if ($null -eq $release) { + $notes = @( + "dlp-io $version", + "", + "## 产物", + "", + "- ``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.tar.gz``：sdist 源码包", + "- ``dlp_io.py`` / ``dlp_io.cp3XX-win_amd64.pyd``：免安装单文件，直接放进项目目录即可 ``import dlp_io``", + "", + "## 安装", + "", + "纯 Python wheel（跨平台）：", + "", + "``````", + "py -m pip install $env:GITEA_SERVER/$env:GITEA_REPOSITORY/releases/download/$tag/dlp_io-$version-py3-none-any.whl", + "``````", + "", + "pyd wheel（Windows，示例为 CPython 3.13，其它版本替换文件名中的 cp313）：", + "", + "``````", + "py -3.13 -m pip install $env:GITEA_SERVER/$env:GITEA_REPOSITORY/releases/download/$tag/dlp_io-$version-cp313-cp313-win_amd64.whl", + "``````" + ) -join "`n" + $body = @{ + tag_name = $tag + name = "dlp-io $version" + body = $notes + draft = $false + prerelease = $false + } | ConvertTo-Json + $release = Invoke-RestMethod -Method Post -Headers $headers -Uri "$api/releases" -Body $body -ContentType "application/json" + Write-Host "created release: $tag" + } else { + Write-Host "release already exists: $tag" + } + + 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)" + } + 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)" + } + + $published = Invoke-RestMethod -Headers $headers -Uri "$api/releases/$($release.id)" + $publishedNames = @($published.assets | ForEach-Object name) + $localNames = @($artifacts | ForEach-Object Name) + if (Compare-Object $localNames $publishedNames) { + throw "release asset mismatch: $($publishedNames -join ', ')" + } + Write-Host "release verified: $tag ($($localNames.Count) files)" diff --git a/MANIFEST.in b/MANIFEST.in index 3b000f1..b44539d 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1,6 +1,5 @@ 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 index 1c0adfc..89c1582 100644 --- a/README.md +++ b/README.md @@ -13,8 +13,24 @@ pytest ## 构建发布 +发布流程由 `.gitea/workflows/publish-dlp-io.yaml` 驱动：推送 `dlp-io-vX.Y.Z` 格式的 annotated tag，Windows runner（Python 3.10–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.tar.gz`（sdist） +- `dlp_io.py` 和 5 个裸 `dlp_io.cp3XX-win_amd64.pyd`（免安装单文件，放进项目目录即可 import） + +安装方式见 [docs/DLP_IO_LIBRARY.md](docs/DLP_IO_LIBRARY.md)。 + +本地只构建纯 Python 版： + ```bash py -3.13 -m build ``` -发布流程由 `.gitea/workflows/publish-dlp-io.yaml` 驱动：推送 `dlp-io-vX.Y.Z` 格式的 annotated tag 即可触发测试、构建、上传到 Gitea PyPI 仓库。 +pyd wheel 由 CI 编译，不需要本地编译；runner 主机需预装 MSVC 生成工具（Visual Studio Build Tools 的 C++ 工作负载）。如需本地验证 pyd 构建，安装 `cython` 后执行： + +```bash +set DLP_IO_PYD=1 +py -3.13 -m build --wheel --no-isolation +``` diff --git a/dlp_io.py b/dlp_io.py new file mode 100644 index 0000000..b61c9aa --- /dev/null +++ b/dlp_io.py @@ -0,0 +1,903 @@ +"""io-compatible file reads through an approved Python helper on Windows DLP hosts.""" + +from __future__ import annotations + +import atexit +import builtins +import io +import json +import operator +import os +import queue +import shutil +import subprocess +import threading +from dataclasses import dataclass +from enum import Enum +from multiprocessing.context import AuthenticationError +from multiprocessing.connection import Client + +__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", +] + + +# --------------------------------------------------------------------------- +# Errors + + +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.""" + + +# --------------------------------------------------------------------------- +# Configuration + + +@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] + + +# --------------------------------------------------------------------------- +# Helper protocol + +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 + + +# --------------------------------------------------------------------------- +# Pipe-backed raw stream + + +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() + + +# --------------------------------------------------------------------------- +# Helper session + +_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, + ): + 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 + + +# --------------------------------------------------------------------------- +# Public io-compatible API + +_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) + + +# --------------------------------------------------------------------------- +# Global open() patch + +_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 = open + io.open = 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/__init__.py b/dlp_io/__init__.py deleted file mode 100644 index 3eb4a38..0000000 --- a/dlp_io/__init__.py +++ /dev/null @@ -1,31 +0,0 @@ -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 deleted file mode 100644 index 85b10be..0000000 --- a/dlp_io/_api.py +++ /dev/null @@ -1,144 +0,0 @@ -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 deleted file mode 100644 index 775609a..0000000 --- a/dlp_io/_config.py +++ /dev/null @@ -1,14 +0,0 @@ -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 deleted file mode 100644 index 8e9eed7..0000000 --- a/dlp_io/_errors.py +++ /dev/null @@ -1,22 +0,0 @@ -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 deleted file mode 100644 index 202c115..0000000 --- a/dlp_io/_helper.py +++ /dev/null @@ -1,172 +0,0 @@ -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 deleted file mode 100644 index 3f0aa2a..0000000 --- a/dlp_io/_patch.py +++ /dev/null @@ -1,94 +0,0 @@ -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 deleted file mode 100644 index 2574507..0000000 --- a/dlp_io/_raw.py +++ /dev/null @@ -1,134 +0,0 @@ -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 deleted file mode 100644 index 0b72818..0000000 --- a/dlp_io/_session.py +++ /dev/null @@ -1,319 +0,0 @@ -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 deleted file mode 100644 index e69de29..0000000 diff --git a/docs/DLP_IO_LIBRARY.md b/docs/DLP_IO_LIBRARY.md index 9521197..7118094 100644 --- a/docs/DLP_IO_LIBRARY.md +++ b/docs/DLP_IO_LIBRARY.md @@ -6,17 +6,28 @@ ## 安装 -当前 Gitea Python Package Registry 已开放匿名下载。Package 页面由 Gitea 根据上传后的 metadata 自动生成；本项目的 `.gitea/workflows/publish-dlp-io.yaml` 负责从 immutable tag 构建 wheel/sdist 并通过 `twine upload` 发布，不生成静态网页。 +发布物托管在 Gitea Release 页面，由 `.gitea/workflows/publish-dlp-io.yaml` 从 immutable tag `dlp-io-vX.Y.Z` 自动构建并上传，全部产物在 CI runner 上编译，不依赖本地环境。 -Package 页面： +Release 页面： -Windows PowerShell： +每个版本包含以下文件： + +- `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.tar.gz`：sdist 源码包。 +- `dlp_io.py` 与 `dlp_io.cp310-win_amd64.pyd` 至 `cp314`：免安装单文件，与 wheel 内容一致；直接放进项目目录（或加入 `PYTHONPATH`）即可 `import dlp_io`，适合不便使用 pip 的环境。 + +按解释器版本选择对应资产，可直接用 pip 安装 URL。Windows PowerShell： ```powershell -py -m pip install --index-url https://gitea.docker.antior.cn/api/packages/antior/pypi/simple dlp-io==0.1.0 +# 纯 Python 版（任意平台） +py -m pip install https://gitea.docker.antior.cn/antior/dlp-io/releases/download/dlp-io-v0.1.0/dlp_io-0.1.0-py3-none-any.whl + +# pyd 版（示例为 CPython 3.13，其它版本替换文件名中的 cp313） +py -3.13 -m pip install https://gitea.docker.antior.cn/antior/dlp-io/releases/download/dlp-io-v0.1.0/dlp_io-0.1.0-cp313-cp313-win_amd64.whl ``` -安装和下载不需要 token。上传新版本、覆盖管理和删除 Package 仍需 Gitea 凭据；发布凭据只能放在 workflow secret 中，不得写入命令行、URL 或 tracked 文件。 +也可以先从 Release 页面下载 wheel，再 `py -m pip install <文件路径>`。下载 Release 资产不需要 token；发布只能在 CI 中通过注入的 `GITHUB_TOKEN` 完成，发布凭据不得写入命令行、URL 或 tracked 文件。 调用方需要安装 `dlp-io`。白名单 Python 无需安装该包：helper 源码由调用方以 UTF-8 通过 stdin 注入，且只依赖 Python 标准库，不会在磁盘创建临时 `.py` 文件。 diff --git a/pyproject.toml b/pyproject.toml index 3106dbf..99a4632 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,27 +7,20 @@ 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" +requires-python = ">=3.10" 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", + "Programming Language :: Python :: 3.14", ] [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/requirements-dev.txt b/requirements-dev.txt index 1978f20..bfc1f53 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,7 +1,9 @@ build +cython packaging pytest pytest-cov +pyyaml setuptools>=68,<82 tomli; python_version < "3.11" wheel diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..2117e84 --- /dev/null +++ b/setup.py @@ -0,0 +1,33 @@ +"""Build hook for dlp-io. + +Default builds (``python -m build``) stay pure Python and produce the +``py3-none-any`` wheel declared in ``pyproject.toml``. + +Setting ``DLP_IO_PYD=1`` compiles the single ``dlp_io`` module into a C +extension with Cython, producing a platform wheel (for example +``cp313-cp313-win_amd64``) that ships one ``.pyd`` binary instead of +the ``.py`` source. +""" + +from __future__ import annotations + +import os + +from setuptools import setup + +_BUILD_PYD = os.environ.get("DLP_IO_PYD") == "1" +_MODULE = "dlp_io" + +if _BUILD_PYD: + from Cython.Build import cythonize + + setup( + ext_modules=cythonize( + [f"{_MODULE}.py"], + build_dir=os.path.join("build", "cython"), + compiler_directives={"language_level": "3"}, + ), + py_modules=[], + ) +else: + setup(py_modules=[_MODULE]) diff --git a/tests/test_dlp_io_api.py b/tests/test_dlp_io_api.py index 1d5b387..d5353ae 100644 --- a/tests/test_dlp_io_api.py +++ b/tests/test_dlp_io_api.py @@ -8,7 +8,7 @@ from pathlib import Path import pytest import dlp_io -import dlp_io._api as api_module +import dlp_io as api_module class _FakeSession: diff --git a/tests/test_dlp_io_packaging.py b/tests/test_dlp_io_packaging.py index 0a56244..e92df39 100644 --- a/tests/test_dlp_io_packaging.py +++ b/tests/test_dlp_io_packaging.py @@ -11,6 +11,7 @@ from enum import Enum from pathlib import Path, PurePosixPath import pytest +import yaml from packaging.specifiers import SpecifierSet from packaging.utils import canonicalize_name, parse_wheel_filename from packaging.version import Version @@ -19,11 +20,12 @@ import dlp_io try: import tomllib -except ModuleNotFoundError: # Python 3.9-3.10 +except ModuleNotFoundError: # Python 3.10 import tomli as tomllib PROJECT_ROOT = Path(__file__).resolve().parents[1] +WORKFLOW_PATH = PROJECT_ROOT / ".gitea" / "workflows" / "publish-dlp-io.yaml" class DynamicField(Enum): @@ -38,6 +40,14 @@ class PackageClassifier(Enum): WINDOWS = "Operating System :: Microsoft :: Windows" +class WorkflowStep(Enum): + CLONE_TAG = "Clone immutable package tag" + TEST_MATRIX = "Test Python 3.10-3.14" + BUILD = "Build and check distribution" + SMOKE = "Smoke test release artifacts" + RELEASE = "Create Gitea release with wheels" + + @dataclass(frozen=True) class BuiltDistributions: wheel: Path @@ -49,15 +59,25 @@ def _load_pyproject() -> dict: return tomllib.load(file) +def _load_workflow() -> dict: + with WORKFLOW_PATH.open(encoding="utf-8") as file: + return yaml.safe_load(file) + + +def _copy_source_tree(source: Path) -> None: + source.mkdir() + shutil.copy2(PROJECT_ROOT / "pyproject.toml", source) + shutil.copy2(PROJECT_ROOT / "MANIFEST.in", source) + shutil.copy2(PROJECT_ROOT / "setup.py", source) + shutil.copy2(PROJECT_ROOT / "dlp_io.py", source) + shutil.copytree(PROJECT_ROOT / "docs", source / "docs") + + @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") + _copy_source_tree(source) dist = root / "dist" completed = subprocess.run( @@ -85,10 +105,9 @@ def built_distributions(tmp_path_factory: pytest.TempPathFactory) -> BuiltDistri 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 SpecifierSet(project["requires-python"]) == SpecifierSet(">=3.10") assert {DynamicField(item) for item in project["dynamic"]} == { DynamicField.VERSION } @@ -99,10 +118,6 @@ def test_pyproject_declares_structured_public_package_contract() -> None: 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) @@ -121,8 +136,8 @@ def test_wheel_contains_only_installable_dlp_io_runtime( 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 + assert PurePosixPath("dlp_io.py") in members + assert not [item for item in members if item.suffix == ".pyd"] member_roots = {Path(item.parts[0]) for item in members} assert member_roots.isdisjoint({Path("tests"), Path("Lib")}) assert PurePosixPath(".env") not in members @@ -139,9 +154,23 @@ def test_sdist_contains_package_sources_and_public_readme_only( } assert PurePosixPath("pyproject.toml") in members + assert PurePosixPath("setup.py") in members assert PurePosixPath("docs/DLP_IO_LIBRARY.md") in members - assert PurePosixPath("dlp_io/__init__.py") in members + assert PurePosixPath("dlp_io.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 + + +def test_publish_workflow_tests_builds_and_releases_tagged_distributions() -> None: + document = _load_workflow() + trigger = document.get("on", document.get(True)) + assert set(trigger["push"]["tags"]) == {"dlp-io-v*.*.*"} + assert "workflow_dispatch" in trigger + + job = document["jobs"]["publish"] + assert job["runs-on"] == "windows-latest" + assert Path(job["defaults"]["run"]["working-directory"]) == Path("repo") + step_names = [step["name"] for step in job["steps"]] + assert step_names == [step.value for step in WorkflowStep] diff --git a/tests/test_dlp_io_session.py b/tests/test_dlp_io_session.py index a942664..e10e523 100644 --- a/tests/test_dlp_io_session.py +++ b/tests/test_dlp_io_session.py @@ -18,14 +18,14 @@ from dlp_io import ( DlpSessionBusyError, DlpTransportError, ) -import dlp_io._session as session_module -from dlp_io._helper import ( +import dlp_io as session_module +from dlp_io import ( CHUNK_SIZE, HelperErrorCode, HelperSourceConfig, render_helper_source, ) -from dlp_io._raw import _PipeRawIO +from dlp_io import _PipeRawIO pytestmark = pytest.mark.skipif(os.name != "nt", reason="AF_PIPE requires Windows")