thegame_1980/TwinVine
1
0
Fork 0
mirror of https://github.com/vinefeeder/TwinVine.git synced 2026-07-15 18:10:04 +02:00
Code Issues Packages Projects Releases Wiki Activity

updates

Browse Source
This commit is contained in:
VineFeeder
2026-01-15 12:15:14 +00:00
parent 377ba82545
commit 8f61d97bbd
22 changed files with 2217 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/envied/src/Cookies/COOKIES.txt
+3
View File
@@ -0,0 +1,3 @@
Add you cookies files to this folder.
Name each file to match the service eg DNSP.txt
packages/envied/src/envied/CHANGELOG.md
+223 -2
View File
@@ -5,6 +5,225 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [2.2.0] - 2026-01-15
### Added
- **CDM-Aware PlayReady Fallback Detection**: Intelligent DRM fallback based on selected CDM
- Adds PlayReady PSSH/KID extraction from track and init data with CDM-aware ordering
- When PlayReady CDM is selected, tries PlayReady first then falls back to Widevine
- When Widevine CDM is selected (default), tries Widevine first then falls back to PlayReady
- **Comprehensive Debug Logging**: Enhanced debug logging for downloaders and muxing
- Added detailed debug logging to aria2c, curl_impersonate, n_m3u8dl_re, and requests downloaders
- Enhanced manifest parsers (DASH, HLS, ISM) with debug logging
- Added debug logging to track muxing operations
### Fixed
- **Hybrid DV+HDR10 Filename Detection**: Fixed HDR10 detection in hybrid Dolby Vision filenames
- Hybrid DV+HDR10 files were incorrectly named "DV.H.265" instead of "DV.HDR.H.265"
- Now checks both `hdr_format_full` and `hdr_format_commercial` fields for HDR10 indicators
- **Vault Adaptive Batch Sizing**: Improved bulk key operations with adaptive batch sizing
- Prevents query limit issues when retrieving large numbers of keys from vaults
- Dynamically adjusts batch sizes based on vault response characteristics
- **Test Command Improvements**: Enhanced test command error detection and sorting
- Improved error detection in test command output
- Added natural sorting for test results
## [2.1.0] - 2025-11-27
### Added
- **Per-Track Quality-Based CDM Selection**: Dynamic CDM switching during runtime DRM operations
- Enables quality-based CDM selection during runtime DRM switching
- Different CDMs can be used for different video quality levels within the same download session
- Example: Use Widevine L3 for SD/HD and PlayReady SL3 for 4K content
- **Enhanced Track Export**: Improved export functionality with additional metadata
- Added URL field to track export for easier identification
- Added descriptor information in export output
- Keys now exported in hex-formatted strings
### Changed
- **Dependencies**: Upgraded to latest compatible versions
- Updated various dependencies to their latest versions
### Fixed
- **Attachment Preservation**: Fixed attachments being dropped during track filtering
- Attachments (screenshots, fonts) were being lost when track list was rebuilt
- Fixes image files remaining in temp directory after muxing
- **DASH BaseURL Resolution**: Added AdaptationSet-level BaseURL support per DASH spec
- URL resolution chain now properly follows: MPD → Period → AdaptationSet → Representation
- **WindscribeVPN Region Support**: Restricted to supported regions with proper error handling
- Added error handling for unsupported regions in get_proxy method
- Prevents cryptic errors when using unsupported region codes
- **Filename Sanitization**: Fixed space-hyphen-space handling in filenames
- Pre-process space-hyphen-space patterns (e.g., "Title - Episode") before other replacements
- Made space-hyphen-space handling conditional on scene_naming setting
- Addresses PR #44 by fixing the root cause
- **CICP Enum Values**: Corrected values to match ITU-T H.273 specification
- Added Primaries.Unspecified (value 2) per H.273 spec
- Renamed Primaries/Transfer value 0 from Unspecified to Reserved for spec accuracy
- Simplified Transfer value 2 from Unspecified_Image to Unspecified
- Verified against ITU-T H.273, ISO/IEC 23091-2, H.264/H.265 specs, and FFmpeg enums
- **HLS Byte Range Parsing**: Fixed TypeError in range_offset conversion
- Converted range_offset to int to prevent TypeError in calculate_byte_range
- **pyplayready Compatibility**: Pinned to <0.7 to avoid KID extraction bug
## [2.0.0] - 2025-11-10
### Breaking Changes
- **REST API Integration**: Core architecture modified to support REST API functionality
- Changes to internal APIs for download management and tracking
- Title and track classes updated with API integration points
- Core component interfaces modified for queue management support
- **Configuration Changes**: New required configuration options for API and enhanced features
- Added `simkl_client_id` now required for Simkl functionality
- Service-specific configuration override structure introduced
- Debug logging configuration options added
- **Forced Subtitles**: Behavior change for forced subtitle inclusion
- Forced subs no longer auto-included, requires explicit `--forced-subs` flag
### Added
- **REST API Server**: Complete download management via REST API (early development)
- Implemented download queue management and worker system
- Added OpenAPI/Swagger documentation for easy API exploration
- Included download progress tracking and status endpoints
- API authentication and comprehensive error handling
- Updated core components to support API integration
- Early development work with more changes planned
- **CustomRemoteCDM**: Highly configurable custom CDM API support
- Support for third-party CDM API providers with maximum configurability
- Full configuration through YAML without code changes
- Addresses GitHub issue #26 for flexible CDM integration
- **WindscribeVPN Proxy Provider**: New VPN provider support
- Added WindscribeVPN following NordVPN and SurfsharkVPN patterns
- Fixes GitHub issue #29
- **Latest Episode Download**: New `--latest-episode` CLI option
- `-le, --latest-episode` flag to download only the most recent episode
- Automatically selects the single most recent episode regardless of season
- Fixes GitHub issue #28
- **Video Track Exclusion**: New `--no-video` CLI option
- `-nv, --no-video` flag to skip downloading video tracks
- Allows downloading only audio, subtitles, attachments, and chapters
- Useful for audio-only or subtitle extraction workflows
- Fixes GitHub issue #39
- **Service-Specific Configuration Overrides**: Per-service fine-tuned control
- Support for per-service configuration overrides in YAML
- Fine-tuned control of downloader and command options per service
- Fixes GitHub issue #13
- **Comprehensive JSON Debug Logging**: Structured logging for troubleshooting
- Binary toggle via `--debug` flag or `debug: true` in config
- JSON Lines (.jsonl) format for easy parsing and analysis
- Comprehensive logging of all operations (session info, CLI params, CDM details, auth status, title/track metadata, DRM operations, vault queries)
- Configurable key logging via `debug_keys` option with smart redaction
- Error logging for all critical operations
- Removed old text logging system
- **curl_cffi Retry Handler**: Enhanced session reliability
- Added automatic retry mechanism to curl_cffi Session
- Improved download reliability with configurable retries
- **Simkl API Configuration**: New API key support
- Added `simkl_client_id` configuration option
- Simkl now requires client_id from https://simkl.com/settings/developer/
- **Custom Session Fingerprints**: Enhanced browser impersonation capabilities
- Added custom fingerprint and preset support for better service compatibility
- Configurable fingerprint presets for different device types
- Improved success rate with services using advanced bot detection
- **TMDB and Simkl Metadata Caching**: Enhanced title cache system
- Added metadata caching to title cache to reduce API calls
- Caches movie/show metadata alongside title information
- Improves performance for repeated title lookups and reduces API rate limiting
- **API Enhancements**: Improved REST API functionality
- Added default parameter handling for better request processing
- Added URL field to services endpoint response for easier service identification
- Complete API enhancements for production readiness
- Improved error responses with better detail and debugging information
### Changed
- **Binary Search Enhancement**: Improved binary discovery
- Refactored to search for binaries in root of binary folder or subfolder named after the binary
- Better organization of binary dependencies
- **Type Annotations**: Modernized to PEP 604 syntax
- Updated session.py type annotations to use modern Python syntax
- Improved code readability and type checking
### Fixed
- **Audio Description Track Support**: Added option to download audio description tracks
- Added `--audio-description/-ad` flag to optionally include descriptive audio tracks
- Previously, audio description tracks were always filtered out
- Users can now choose to download AD tracks when needed
- Fixes GitHub issue #33
- **Config Directory Support**: Cross-platform user config directory support
- Fixed config loading to properly support user config directories across all platforms
- Fixes GitHub issue #23
- **HYBRID Mode Validation**: Pre-download validation for hybrid processing
- Added validation to check both HDR10 and DV tracks are available before download
- Prevents wasted downloads when hybrid processing would fail
- **TMDB/Simkl API Keys**: Graceful handling of missing API keys
- Improved error handling when TMDB or Simkl API keys are not configured
- Better user messaging for API configuration requirements
- **Forced Subtitles Behavior**: Correct forced subtitle filtering
- Fixed forced subtitles being incorrectly included without `--forced-subs` flag
- Forced subs now only included when explicitly requested
- **Font Attachment Constructor**: Fixed ASS/SSA font attachment
- Use keyword arguments for Attachment constructor in font attachment
- Fixes "Invalid URL: No scheme supplied" error
- Fixes GitHub issue #24
- **Binary Subdirectory Checking**: Enhanced binary location discovery (by @TPD94, PR #19)
- Updated binaries.py to check subdirectories in binaries folders named after the binary
- Improved binary detection and loading
- **HLS Manifest Processing**: Minor HLS parser fix (by @TPD94, PR #19)
- **lxml and pyplayready**: Updated dependencies (by @Sp5rky)
- Updated lxml constraint and pyplayready import path for compatibility
- **DASH Segment Calculation**: Corrected segment handling
- Fixed segment count calculation for DASH manifests with startNumber=0
- Ensures accurate segment processing for all DASH manifest configurations
- Prevents off-by-one errors in segment downloads
- **HDR Detection and Naming**: Comprehensive HDR format support
- Improved HDR detection with comprehensive transfer characteristics checks
- Added hybrid DV+HDR10 support for accurate file naming
- Better identification of HDR formats across different streaming services
- More accurate HDR/DV detection in filename generation
- **Subtitle Processing**: VTT subtitle handling improvements
- Resolved SDH (Subtitles for Deaf and Hard of hearing) stripping crash when processing VTT files
- More robust subtitle processing pipeline with better error handling
- Fixes crashes when filtering specific VTT subtitle formats
- **DRM Processing**: Enhanced encoding handling
- Added explicit UTF-8 encoding to mp4decrypt subprocess calls
- Prevents encoding issues on systems with non-UTF-8 default encodings
- Improves cross-platform compatibility for Windows and some Linux configurations
- **Session Fingerprints**: Updated OkHttp presets
- Updated OkHttp fingerprint presets for better Android TV compatibility
- Improved success rate with services using fingerprint-based detection
### Documentation
- **GitHub Issue Templates**: Enhanced issue reporting
- Improved bug report template with better structure and required fields
- Enhanced feature request template for clearer specifications
- Added helpful guidance for contributors to provide complete information
### Refactored
- **Import Cleanup**: Removed unused imports
- Removed unused mypy import from binaries.py
- Fixed import ordering in API download_manager and handlers
### Contributors
This release includes contributions from:
- @Sp5rky - REST API server implementation, dependency updates
- @stabbedbybrick - curl_cffi retry handler (PR #31)
- @stabbedbybrick - n_m3u8dl-re refactor (PR #38)
- @TPD94 - Binary search enhancements, manifest parser fixes (PR #19)
- @scene (Andy) - Core features, configuration system, bug fixes
## [1.4.8] - 2025-10-08
### Added
@@ -13,6 +232,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Enables strict language code matching without fallbacks
- **No-Mux Flag**: New `--no-mux` flag to skip muxing tracks into container files
- Useful for keeping individual track files separate
- **DecryptLabs API Integration for HTTP Vault**: Enhanced vault support
- Added DecryptLabs API support to HTTP vault for improved key retrieval
- **AC4 Audio Codec Support**: Enhanced audio format handling
- Added AC4 codec support in Audio class with updated mime/profile handling
- **pysubs2 Subtitle Conversion**: Extended subtitle format support
@@ -177,7 +398,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
- **Matroska Tag Compliance**: Enhanced media container compatibility
- **Matroska Tag Compliance**: Enhanced media container compatibility
- Fixed Matroska tag compliance with official specification
- **Application Branding**: Cleaned up version display
- Removed old devine version reference from banner to avoid developer confusion
@@ -456,4 +677,4 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Updated package version from 1.0.2 to 1.1.0
- Enhanced dl.py command processing for hybrid mode support
- Improved core titles (episode/movie) processing for HDR content
- Extended tracks module with hybrid processing capabilities
- Extended tracks module with hybrid processing capabilities
packages/envied/src/envied/commands/util.py
+14 -6
View File
@@ -1,3 +1,4 @@
import re
import subprocess
from pathlib import Path
@@ -8,6 +9,11 @@ from envied.core import binaries
from envied.core.constants import context_settings
def _natural_sort_key(path: Path) -> list:
"""Sort key for natural sorting (S01E01 before S01E10)."""
return [int(part) if part.isdigit() else part.lower() for part in re.split(r"(\d+)", path.name)]
@click.group(short_help="Various helper scripts and programs.", context_settings=context_settings)
def util() -> None:
"""Various helper scripts and programs."""
@@ -49,7 +55,7 @@ def crop(path: Path, aspect: str, letter: bool, offset: int, preview: bool) -> N
raise click.ClickException('FFmpeg executable "ffmpeg" not found but is required.')
if path.is_dir():
paths = list(path.glob("*.mkv")) + list(path.glob("*.mp4"))
paths = sorted(list(path.glob("*.mkv")) + list(path.glob("*.mp4")), key=_natural_sort_key)
else:
paths = [path]
for video_path in paths:
@@ -140,7 +146,7 @@ def range_(path: Path, full: bool, preview: bool) -> None:
raise click.ClickException('FFmpeg executable "ffmpeg" not found but is required.')
if path.is_dir():
paths = list(path.glob("*.mkv")) + list(path.glob("*.mp4"))
paths = sorted(list(path.glob("*.mkv")) + list(path.glob("*.mp4")), key=_natural_sort_key)
else:
paths = [path]
for video_path in paths:
@@ -225,16 +231,18 @@ def test(path: Path, map_: str) -> None:
raise click.ClickException('FFmpeg executable "ffmpeg" not found but is required.')
if path.is_dir():
paths = list(path.glob("*.mkv")) + list(path.glob("*.mp4"))
paths = sorted(list(path.glob("*.mkv")) + list(path.glob("*.mp4")), key=_natural_sort_key)
else:
paths = [path]
for video_path in paths:
print("Starting...")
print(f"Testing: {video_path.name}")
p = subprocess.Popen(
[
binaries.FFMPEG,
"-hide_banner",
"-benchmark",
"-err_detect",
"+crccheck+bitstream+buffer+careful+compliant+aggressive",
"-i",
str(video_path),
"-map",
@@ -255,13 +263,13 @@ def test(path: Path, map_: str) -> None:
reached_output = True
if not reached_output:
continue
if line.startswith("["): # error of some kind
if line.startswith("[") and not line.startswith("[out#"):
errors += 1
stream, error = line.split("] ", maxsplit=1)
stream = stream.split(" @ ")[0]
line = f"{stream} ERROR: {error}"
print(line)
p.stderr.close()
print(f"Finished with {errors} Errors, Cleaning up...")
print(f"Finished with {errors} error(s)")
p.terminate()
p.wait()
packages/envied/src/envied/core/__init__.py
+1 -1
View File
@@ -1 +1 @@
__version__ = "2.1.0"
__version__ = "2.2.0"
packages/envied/src/envied/core/downloaders/aria2c.py
+82 -2
View File
@@ -19,7 +19,7 @@ from envied.core import binaries
from envied.core.config import config
from envied.core.console import console
from envied.core.constants import DOWNLOAD_CANCELLED
from envied.core.utilities import get_extension, get_free_port
from envied.core.utilities import get_debug_logger, get_extension, get_free_port
def rpc(caller: Callable, secret: str, method: str, params: Optional[list[Any]] = None) -> Any:
@@ -58,6 +58,8 @@ def download(
proxy: Optional[str] = None,
max_workers: Optional[int] = None,
) -> Generator[dict[str, Any], None, None]:
debug_logger = get_debug_logger()
if not urls:
raise ValueError("urls must be provided and not empty")
elif not isinstance(urls, (str, dict, list)):
@@ -91,6 +93,13 @@ def download(
urls = [urls]
if not binaries.Aria2:
if debug_logger:
debug_logger.log(
level="ERROR",
operation="downloader_aria2c_binary_missing",
message="Aria2c executable not found in PATH or local binaries directory",
context={"searched_names": ["aria2c", "aria2"]},
)
raise EnvironmentError("Aria2c executable not found...")
if proxy and not proxy.lower().startswith("http://"):
@@ -180,6 +189,28 @@ def download(
continue
arguments.extend(["--header", f"{header}: {value}"])
if debug_logger:
first_url = urls[0] if isinstance(urls[0], str) else urls[0].get("url", "")
url_display = first_url[:200] + "..." if len(first_url) > 200 else first_url
debug_logger.log(
level="DEBUG",
operation="downloader_aria2c_start",
message="Starting Aria2c download",
context={
"binary_path": str(binaries.Aria2),
"url_count": len(urls),
"first_url": url_display,
"output_dir": str(output_dir),
"filename": filename,
"max_concurrent_downloads": max_concurrent_downloads,
"max_connection_per_server": max_connection_per_server,
"split": split,
"file_allocation": file_allocation,
"has_proxy": bool(proxy),
"rpc_port": rpc_port,
},
)
yield dict(total=len(urls))
try:
@@ -226,6 +257,20 @@ def download(
textwrap.wrap(error, width=console.width - 20, initial_indent="")
)
console.log(Text.from_ansi("\n[Aria2c]: " + error_pretty))
if debug_logger:
debug_logger.log(
level="ERROR",
operation="downloader_aria2c_download_error",
message=f"Aria2c download failed: {dl['errorMessage']}",
context={
"gid": dl["gid"],
"error_code": dl["errorCode"],
"error_message": dl["errorMessage"],
"used_uri": used_uri[:200] + "..." if len(used_uri) > 200 else used_uri,
"completed_length": dl.get("completedLength"),
"total_length": dl.get("totalLength"),
},
)
raise ValueError(error)
if number_stopped == len(urls):
@@ -237,7 +282,31 @@ def download(
p.wait()
if p.returncode != 0:
if debug_logger:
debug_logger.log(
level="ERROR",
operation="downloader_aria2c_failed",
message=f"Aria2c exited with code {p.returncode}",
context={
"returncode": p.returncode,
"url_count": len(urls),
"output_dir": str(output_dir),
},
)
raise subprocess.CalledProcessError(p.returncode, arguments)
if debug_logger:
debug_logger.log(
level="DEBUG",
operation="downloader_aria2c_complete",
message="Aria2c download completed successfully",
context={
"url_count": len(urls),
"output_dir": str(output_dir),
"filename": filename,
},
)
except ConnectionResetError:
# interrupted while passing URI to download
raise KeyboardInterrupt()
@@ -251,9 +320,20 @@ def download(
DOWNLOAD_CANCELLED.set() # skip pending track downloads
yield dict(downloaded="[yellow]CANCELLED")
raise
except Exception:
except Exception as e:
DOWNLOAD_CANCELLED.set() # skip pending track downloads
yield dict(downloaded="[red]FAILED")
if debug_logger and not isinstance(e, (subprocess.CalledProcessError, ValueError)):
debug_logger.log(
level="ERROR",
operation="downloader_aria2c_exception",
message=f"Unexpected error during Aria2c download: {e}",
error=e,
context={
"url_count": len(urls),
"output_dir": str(output_dir),
},
)
raise
finally:
rpc(caller=partial(rpc_session.post, url=rpc_uri), secret=rpc_secret, method="aria2.shutdown")
packages/envied/src/envied/core/downloaders/curl_impersonate.py
+46 -2
View File
@@ -11,7 +11,7 @@ from rich import filesize
from envied.core.config import config
from envied.core.constants import DOWNLOAD_CANCELLED
from envied.core.utilities import get_extension
from envied.core.utilities import get_debug_logger, get_extension
MAX_ATTEMPTS = 5
RETRY_WAIT = 2
@@ -189,6 +189,8 @@ def curl_impersonate(
if not isinstance(max_workers, (int, type(None))):
raise TypeError(f"Expected max_workers to be {int}, not {type(max_workers)}")
debug_logger = get_debug_logger()
if not isinstance(urls, list):
urls = [urls]
@@ -209,6 +211,24 @@ def curl_impersonate(
if proxy:
session.proxies.update({"all": proxy})
if debug_logger:
first_url = urls[0].get("url", "") if urls else ""
url_display = first_url[:200] + "..." if len(first_url) > 200 else first_url
debug_logger.log(
level="DEBUG",
operation="downloader_curl_impersonate_start",
message="Starting curl_impersonate download",
context={
"url_count": len(urls),
"first_url": url_display,
"output_dir": str(output_dir),
"filename": filename,
"max_workers": max_workers,
"browser": BROWSER,
"has_proxy": bool(proxy),
},
)
yield dict(total=len(urls))
download_sizes = []
@@ -235,11 +255,23 @@ def curl_impersonate(
# tell dl that it was cancelled
# the pool is already shut down, so exiting loop is fine
raise
except Exception:
except Exception as e:
DOWNLOAD_CANCELLED.set() # skip pending track downloads
yield dict(downloaded="[red]FAILING")
pool.shutdown(wait=True, cancel_futures=True)
yield dict(downloaded="[red]FAILED")
if debug_logger:
debug_logger.log(
level="ERROR",
operation="downloader_curl_impersonate_failed",
message=f"curl_impersonate download failed: {e}",
error=e,
context={
"url_count": len(urls),
"output_dir": str(output_dir),
"browser": BROWSER,
},
)
# tell dl that it failed
# the pool is already shut down, so exiting loop is fine
raise
@@ -260,5 +292,17 @@ def curl_impersonate(
last_speed_refresh = now
download_sizes.clear()
if debug_logger:
debug_logger.log(
level="DEBUG",
operation="downloader_curl_impersonate_complete",
message="curl_impersonate download completed successfully",
context={
"url_count": len(urls),
"output_dir": str(output_dir),
"filename": filename,
},
)
__all__ = ("curl_impersonate",)
packages/envied/src/envied/core/downloaders/n_m3u8dl_re.py
+96 -3
View File
@@ -13,6 +13,7 @@ from envied.core import binaries
from envied.core.config import config
from envied.core.console import console
from envied.core.constants import DOWNLOAD_CANCELLED
from envied.core.utilities import get_debug_logger
PERCENT_RE = re.compile(r"(\d+\.\d+%)")
SPEED_RE = re.compile(r"(\d+\.\d+(?:MB|KB)ps)")
@@ -176,7 +177,6 @@ def build_download_args(
"--thread-count": thread_count,
"--download-retry-count": retry_count,
"--write-meta-json": False,
"--no-log": True,
}
if proxy:
args["--custom-proxy"] = proxy
@@ -224,6 +224,8 @@ def download(
content_keys: dict[str, Any] | None,
skip_merge: bool | None = False,
) -> Generator[dict[str, Any], None, None]:
debug_logger = get_debug_logger()
if not urls:
raise ValueError("urls must be provided and not empty")
if not isinstance(urls, (str, dict, list)):
@@ -275,7 +277,39 @@ def download(
skip_merge=skip_merge,
ad_keyword=ad_keyword,
)
arguments.extend(get_track_selection_args(track))
selection_args = get_track_selection_args(track)
arguments.extend(selection_args)
log_file_path: Path | None = None
if debug_logger:
log_file_path = output_dir / f".n_m3u8dl_re_{filename}.log"
arguments.extend(["--log-file-path", str(log_file_path)])
track_url_display = track.url[:200] + "..." if len(track.url) > 200 else track.url
debug_logger.log(
level="DEBUG",
operation="downloader_n_m3u8dl_re_start",
message="Starting N_m3u8DL-RE download",
context={
"binary_path": str(binaries.N_m3u8DL_RE),
"track_id": getattr(track, "id", None),
"track_type": track.__class__.__name__,
"track_url": track_url_display,
"output_dir": str(output_dir),
"filename": filename,
"thread_count": thread_count,
"retry_count": retry_count,
"has_content_keys": bool(content_keys),
"content_key_count": len(content_keys) if content_keys else 0,
"has_proxy": bool(proxy),
"skip_merge": skip_merge,
"has_custom_args": bool(track.downloader_args),
"selection_args": selection_args,
"descriptor": track.descriptor.name if hasattr(track, "descriptor") else None,
},
)
else:
arguments.extend(["--no-log", "true"])
yield {"total": 100}
yield {"downloaded": "Parsing streams..."}
@@ -310,11 +344,45 @@ def download(
yield {"completed": progress} if progress < 100 else {"downloaded": "Merging"}
process.wait()
if process.returncode != 0:
if debug_logger and log_file_path:
log_contents = ""
if log_file_path.exists():
try:
log_contents = log_file_path.read_text(encoding="utf-8", errors="replace")
except Exception:
log_contents = "<failed to read log file>"
debug_logger.log(
level="ERROR",
operation="downloader_n_m3u8dl_re_failed",
message=f"N_m3u8DL-RE exited with code {process.returncode}",
context={
"returncode": process.returncode,
"track_id": getattr(track, "id", None),
"track_type": track.__class__.__name__,
"last_line": last_line,
"log_file_contents": log_contents,
},
)
if error_match := ERROR_RE.search(last_line):
raise ValueError(f"[N_m3u8DL-RE]: {error_match.group(1)}")
raise subprocess.CalledProcessError(process.returncode, arguments)
if debug_logger:
debug_logger.log(
level="DEBUG",
operation="downloader_n_m3u8dl_re_complete",
message="N_m3u8DL-RE download completed successfully",
context={
"track_id": getattr(track, "id", None),
"track_type": track.__class__.__name__,
"output_dir": str(output_dir),
"filename": filename,
},
)
except ConnectionResetError:
# interrupted while passing URI to download
raise KeyboardInterrupt()
@@ -322,10 +390,35 @@ def download(
DOWNLOAD_CANCELLED.set() # skip pending track downloads
yield {"downloaded": "[yellow]CANCELLED"}
raise
except Exception:
except Exception as e:
DOWNLOAD_CANCELLED.set() # skip pending track downloads
yield {"downloaded": "[red]FAILED"}
if debug_logger and log_file_path and not isinstance(e, (subprocess.CalledProcessError, ValueError)):
log_contents = ""
if log_file_path.exists():
try:
log_contents = log_file_path.read_text(encoding="utf-8", errors="replace")
except Exception:
log_contents = "<failed to read log file>"
debug_logger.log(
level="ERROR",
operation="downloader_n_m3u8dl_re_exception",
message=f"Unexpected error during N_m3u8DL-RE download: {e}",
error=e,
context={
"track_id": getattr(track, "id", None),
"track_type": track.__class__.__name__,
"log_file_contents": log_contents,
},
)
raise
finally:
if log_file_path and log_file_path.exists():
try:
log_file_path.unlink()
except Exception:
pass
def n_m3u8dl_re(
packages/envied/src/envied/core/downloaders/requests.py
+44 -2
View File
@@ -12,7 +12,7 @@ from requests.adapters import HTTPAdapter
from rich import filesize
from envied.core.constants import DOWNLOAD_CANCELLED
from envied.core.utilities import get_extension
from envied.core.utilities import get_debug_logger, get_extension
MAX_ATTEMPTS = 5
RETRY_WAIT = 2
@@ -215,6 +215,8 @@ def requests(
if not isinstance(max_workers, (int, type(None))):
raise TypeError(f"Expected max_workers to be {int}, not {type(max_workers)}")
debug_logger = get_debug_logger()
if not isinstance(urls, list):
urls = [urls]
@@ -241,6 +243,23 @@ def requests(
if proxy:
session.proxies.update({"all": proxy})
if debug_logger:
first_url = urls[0].get("url", "") if urls else ""
url_display = first_url[:200] + "..." if len(first_url) > 200 else first_url
debug_logger.log(
level="DEBUG",
operation="downloader_requests_start",
message="Starting requests download",
context={
"url_count": len(urls),
"first_url": url_display,
"output_dir": str(output_dir),
"filename": filename,
"max_workers": max_workers,
"has_proxy": bool(proxy),
},
)
yield dict(total=len(urls))
try:
@@ -256,14 +275,37 @@ def requests(
# tell dl that it was cancelled
# the pool is already shut down, so exiting loop is fine
raise
except Exception:
except Exception as e:
DOWNLOAD_CANCELLED.set() # skip pending track downloads
yield dict(downloaded="[red]FAILING")
pool.shutdown(wait=True, cancel_futures=True)
yield dict(downloaded="[red]FAILED")
if debug_logger:
debug_logger.log(
level="ERROR",
operation="downloader_requests_failed",
message=f"Requests download failed: {e}",
error=e,
context={
"url_count": len(urls),
"output_dir": str(output_dir),
},
)
# tell dl that it failed
# the pool is already shut down, so exiting loop is fine
raise
if debug_logger:
debug_logger.log(
level="DEBUG",
operation="downloader_requests_complete",
message="Requests download completed successfully",
context={
"url_count": len(urls),
"output_dir": str(output_dir),
"filename": filename,
},
)
finally:
DOWNLOAD_SIZES.clear()
packages/envied/src/envied/core/manifests/dash.py
+36 -6
View File
@@ -18,6 +18,7 @@ import requests
from curl_cffi.requests import Session as CurlSession
from langcodes import Language, tag_is_valid
from lxml.etree import Element, ElementTree
from pyplayready.cdm import Cdm as PlayReadyCdm
from pyplayready.system.pssh import PSSH as PR_PSSH
from pywidevine.cdm import Cdm as WidevineCdm
from pywidevine.pssh import PSSH
@@ -28,7 +29,7 @@ from envied.core.downloaders import requests as requests_downloader
from envied.core.drm import DRM_T, PlayReady, Widevine
from envied.core.events import events
from envied.core.tracks import Audio, Subtitle, Tracks, Video
from envied.core.utilities import is_close_match, try_ensure_utf8
from envied.core.utilities import get_debug_logger, is_close_match, try_ensure_utf8
from envied.core.utils.xml import load_xml
@@ -466,11 +467,22 @@ class DASH:
track.data["dash"]["segment_durations"] = segment_durations
if not track.drm and isinstance(track, (Video, Audio)):
try:
track.drm = [Widevine.from_init_data(init_data)]
except Widevine.Exceptions.PSSHNotFound:
# it might not have Widevine DRM, or might not have found the PSSH
log.warning("No Widevine PSSH was found for this track, is it DRM free?")
if isinstance(cdm, PlayReadyCdm):
try:
track.drm = [PlayReady.from_init_data(init_data)]
except PlayReady.Exceptions.PSSHNotFound:
try:
track.drm = [Widevine.from_init_data(init_data)]
except Widevine.Exceptions.PSSHNotFound:
log.warning("No PlayReady or Widevine PSSH was found for this track, is it DRM free?")
else:
try:
track.drm = [Widevine.from_init_data(init_data)]
except Widevine.Exceptions.PSSHNotFound:
try:
track.drm = [PlayReady.from_init_data(init_data)]
except PlayReady.Exceptions.PSSHNotFound:
log.warning("No Widevine or PlayReady PSSH was found for this track, is it DRM free?")
if track.drm:
track_kid = track_kid or track.get_key_id(url=segments[0][0], session=session)
@@ -518,6 +530,24 @@ class DASH:
if downloader.__name__ == "n_m3u8dl_re":
downloader_args.update({"filename": track.id, "track": track})
debug_logger = get_debug_logger()
if debug_logger:
debug_logger.log(
level="DEBUG",
operation="manifest_dash_download_start",
message="Starting DASH manifest download",
context={
"track_id": getattr(track, "id", None),
"track_type": track.__class__.__name__,
"total_segments": len(segments),
"downloader": downloader.__name__,
"has_drm": bool(track.drm),
"drm_types": [drm.__class__.__name__ for drm in (track.drm or [])],
"save_path": str(save_path),
"has_init_data": bool(init_data),
},
)
for status_update in downloader(**downloader_args):
file_downloaded = status_update.get("file_downloaded")
if file_downloaded:
packages/envied/src/envied/core/titles/episode.py
+4 -1
View File
@@ -185,7 +185,10 @@ class Episode(Title):
if hdr_format:
if hdr_format_full.startswith("Dolby Vision"):
name += " DV"
if any(indicator in hdr_format_full for indicator in ["HDR10", "SMPTE ST 2086"]):
if any(
indicator in (hdr_format_full + " " + hdr_format)
for indicator in ["HDR10", "SMPTE ST 2086"]
):
name += " HDR"
else:
name += f" {DYNAMIC_RANGE_MAP.get(hdr_format)} "
packages/envied/src/envied/core/titles/movie.py
+4 -1
View File
@@ -136,7 +136,10 @@ class Movie(Title):
if hdr_format:
if hdr_format_full.startswith("Dolby Vision"):
name += " DV"
if any(indicator in hdr_format_full for indicator in ["HDR10", "SMPTE ST 2086"]):
if any(
indicator in (hdr_format_full + " " + hdr_format)
for indicator in ["HDR10", "SMPTE ST 2086"]
):
name += " HDR"
else:
name += f" {DYNAMIC_RANGE_MAP.get(hdr_format)} "
packages/envied/src/envied/core/tracks/track.py
+17 -6
View File
@@ -295,12 +295,23 @@ class Track:
try:
if not self.drm and track_type in ("Video", "Audio"):
# the service might not have explicitly defined the `drm` property
# try find widevine DRM information from the init data of URL
try:
self.drm = [Widevine.from_track(self, session)]
except Widevine.Exceptions.PSSHNotFound:
# it might not have Widevine DRM, or might not have found the PSSH
log.warning("No Widevine PSSH was found for this track, is it DRM free?")
# try find DRM information from the init data of URL based on CDM type
if isinstance(cdm, PlayReadyCdm):
try:
self.drm = [PlayReady.from_track(self, session)]
except PlayReady.Exceptions.PSSHNotFound:
try:
self.drm = [Widevine.from_track(self, session)]
except Widevine.Exceptions.PSSHNotFound:
log.warning("No PlayReady or Widevine PSSH was found for this track, is it DRM free?")
else:
try:
self.drm = [Widevine.from_track(self, session)]
except Widevine.Exceptions.PSSHNotFound:
try:
self.drm = [PlayReady.from_track(self, session)]
except PlayReady.Exceptions.PSSHNotFound:
log.warning("No Widevine or PlayReady PSSH was found for this track, is it DRM free?")
if self.drm:
track_kid = self.get_key_id(session=session)
packages/envied/src/envied/core/tracks/tracks.py
+57 -2
View File
@@ -22,7 +22,7 @@ from envied.core.tracks.chapters import Chapter, Chapters
from envied.core.tracks.subtitle import Subtitle
from envied.core.tracks.track import Track
from envied.core.tracks.video import Video
from envied.core.utilities import is_close_match, sanitize_filename
from envied.core.utilities import get_debug_logger, is_close_match, sanitize_filename
from envied.core.utils.collections import as_list, flatten
@@ -507,6 +507,35 @@ class Tracks:
if not output_path:
raise ValueError("No tracks provided, at least one track must be provided.")
debug_logger = get_debug_logger()
if debug_logger:
debug_logger.log(
level="DEBUG",
operation="mux_start",
message="Starting mkvmerge muxing",
context={
"title": title,
"output_path": str(output_path),
"video_count": len(self.videos),
"audio_count": len(self.audio),
"subtitle_count": len(self.subtitles),
"attachment_count": len(self.attachments),
"has_chapters": bool(self.chapters),
"video_tracks": [
{"id": v.id, "codec": getattr(v, "codec", None), "language": str(v.language)}
for v in self.videos
],
"audio_tracks": [
{"id": a.id, "codec": getattr(a, "codec", None), "language": str(a.language)}
for a in self.audio
],
"subtitle_tracks": [
{"id": s.id, "codec": getattr(s, "codec", None), "language": str(s.language)}
for s in self.subtitles
],
},
)
# let potential failures go to caller, caller should handle
try:
errors = []
@@ -516,7 +545,33 @@ class Tracks:
errors.append(line)
if "progress" in line:
progress(total=100, completed=int(line.strip()[14:-1]))
return output_path, p.wait(), errors
returncode = p.wait()
if debug_logger:
if returncode != 0 or errors:
debug_logger.log(
level="ERROR",
operation="mux_failed",
message=f"mkvmerge exited with code {returncode}",
context={
"returncode": returncode,
"output_path": str(output_path),
"errors": errors,
},
)
else:
debug_logger.log(
level="DEBUG",
operation="mux_complete",
message="mkvmerge muxing completed successfully",
context={
"output_path": str(output_path),
"output_exists": output_path.exists() if output_path else False,
},
)
return output_path, returncode, errors
finally:
if chapters_path:
chapters_path.unlink()
packages/envied/src/envied/vaults/API.py
+223
View File
@@ -0,0 +1,223 @@
from typing import Iterator, Optional, Union
from uuid import UUID
from requests import Session
from envied.core import __version__
from envied.core.vault import Vault
class API(Vault):
"""Key Vault using a simple RESTful HTTP API call."""
def __init__(self, name: str, uri: str, token: str, no_push: bool = False):
super().__init__(name, no_push)
self.uri = uri.rstrip("/")
self.session = Session()
self.session.headers.update({"User-Agent": f"unshackle v{__version__}"})
self.session.headers.update({"Authorization": f"Bearer {token}"})
def get_key(self, kid: Union[UUID, str], service: str) -> Optional[str]:
if isinstance(kid, UUID):
kid = kid.hex
data = self.session.get(
url=f"{self.uri}/{service.lower()}/{kid}", headers={"Accept": "application/json"}
).json()
code = int(data.get("code", 0))
message = data.get("message")
error = {
0: None,
1: Exceptions.AuthRejected,
2: Exceptions.TooManyRequests,
3: Exceptions.ServiceTagInvalid,
4: Exceptions.KeyIdInvalid,
}.get(code, ValueError)
if error:
raise error(f"{message} ({code})")
content_key = data.get("content_key")
if not content_key:
return None
if not isinstance(content_key, str):
raise ValueError(f"Expected {content_key} to be {str}, was {type(content_key)}")
return content_key
def get_keys(self, service: str) -> Iterator[tuple[str, str]]:
page = 1
while True:
data = self.session.get(
url=f"{self.uri}/{service.lower()}",
params={"page": page, "total": 10},
headers={"Accept": "application/json"},
).json()
code = int(data.get("code", 0))
message = data.get("message")
error = {
0: None,
1: Exceptions.AuthRejected,
2: Exceptions.TooManyRequests,
3: Exceptions.PageInvalid,
4: Exceptions.ServiceTagInvalid,
}.get(code, ValueError)
if error:
raise error(f"{message} ({code})")
content_keys = data.get("content_keys")
if content_keys:
if not isinstance(content_keys, dict):
raise ValueError(f"Expected {content_keys} to be {dict}, was {type(content_keys)}")
for key_id, key in content_keys.items():
yield key_id, key
pages = int(data["pages"])
if pages <= page:
break
page += 1
def add_key(self, service: str, kid: Union[UUID, str], key: str) -> bool:
if isinstance(kid, UUID):
kid = kid.hex
data = self.session.post(
url=f"{self.uri}/{service.lower()}/{kid}", json={"content_key": key}, headers={"Accept": "application/json"}
).json()
code = int(data.get("code", 0))
message = data.get("message")
error = {
0: None,
1: Exceptions.AuthRejected,
2: Exceptions.TooManyRequests,
3: Exceptions.ServiceTagInvalid,
4: Exceptions.KeyIdInvalid,
5: Exceptions.ContentKeyInvalid,
}.get(code, ValueError)
if error:
raise error(f"{message} ({code})")
# the kid:key was new to the vault (optional)
added = bool(data.get("added"))
# the key for kid was changed/updated (optional)
updated = bool(data.get("updated"))
return added or updated
def add_keys(self, service: str, kid_keys: dict[Union[UUID, str], str]) -> int:
# Normalize keys
normalized_keys = {str(kid).replace("-", ""): key for kid, key in kid_keys.items()}
kid_list = list(normalized_keys.keys())
if not kid_list:
return 0
# Try batches starting at 500, stepping down by 100 on failure, fallback to 1
batch_size = 500
total_added = 0
i = 0
while i < len(kid_list):
batch_kids = kid_list[i : i + batch_size]
batch_keys = {kid: normalized_keys[kid] for kid in batch_kids}
try:
response = self.session.post(
url=f"{self.uri}/{service.lower()}",
json={"content_keys": batch_keys},
headers={"Accept": "application/json"},
)
# Check for HTTP errors that suggest batch is too large
if response.status_code in (413, 414, 400) and batch_size > 1:
if batch_size > 100:
batch_size -= 100
else:
batch_size = 1
continue
data = response.json()
except Exception:
# JSON decode error or connection issue - try smaller batch
if batch_size > 1:
if batch_size > 100:
batch_size -= 100
else:
batch_size = 1
continue
raise
code = int(data.get("code", 0))
message = data.get("message")
error = {
0: None,
1: Exceptions.AuthRejected,
2: Exceptions.TooManyRequests,
3: Exceptions.ServiceTagInvalid,
4: Exceptions.KeyIdInvalid,
5: Exceptions.ContentKeyInvalid,
}.get(code, ValueError)
if error:
raise error(f"{message} ({code})")
# each kid:key that was new to the vault (optional)
added = int(data.get("added", 0))
# each key for a kid that was changed/updated (optional)
updated = int(data.get("updated", 0))
total_added += added + updated
i += batch_size
return total_added
def get_services(self) -> Iterator[str]:
data = self.session.post(url=self.uri, headers={"Accept": "application/json"}).json()
code = int(data.get("code", 0))
message = data.get("message")
error = {
0: None,
1: Exceptions.AuthRejected,
2: Exceptions.TooManyRequests,
}.get(code, ValueError)
if error:
raise error(f"{message} ({code})")
service_list = data.get("service_list", [])
if not isinstance(service_list, list):
raise ValueError(f"Expected {service_list} to be {list}, was {type(service_list)}")
for service in service_list:
yield service
class Exceptions:
class AuthRejected(Exception):
"""Authentication Error Occurred, is your token valid? Do you have permission to make this call?"""
class TooManyRequests(Exception):
"""Rate Limited; Sent too many requests in a given amount of time."""
class PageInvalid(Exception):
"""Requested page does not exist."""
class ServiceTagInvalid(Exception):
"""The Service Tag is invalid."""
class KeyIdInvalid(Exception):
"""The Key ID is invalid."""
class ContentKeyInvalid(Exception):
"""The Content Key is invalid."""
packages/envied/src/envied/vaults/HTTP.py
+410
View File
@@ -0,0 +1,410 @@
import json
from enum import Enum
from typing import Iterator, Optional, Union
from uuid import UUID
from requests import Session
from envied.core import __version__
from envied.core.vault import Vault
class InsertResult(Enum):
FAILURE = 0
SUCCESS = 1
ALREADY_EXISTS = 2
class HTTP(Vault):
"""
Key Vault using HTTP API with support for multiple API modes.
Supported modes:
- query: Uses GET requests with query parameters
- json: Uses POST requests with JSON payloads
- decrypt_labs: Uses DecryptLabs API format (read-only)
"""
def __init__(
self,
name: str,
host: str,
password: Optional[str] = None,
api_key: Optional[str] = None,
username: Optional[str] = None,
api_mode: str = "query",
no_push: bool = False,
):
"""
Initialize HTTP Vault.
Args:
name: Vault name
host: Host URL
password: Password for query mode or API token for json mode
api_key: API key (alternative to password, used for decrypt_labs mode)
username: Username (required for query mode, ignored for json/decrypt_labs mode)
api_mode: "query" for query parameters, "json" for JSON API, or "decrypt_labs" for DecryptLabs API
no_push: If True, this vault will not receive pushed keys
"""
super().__init__(name, no_push)
self.url = host
self.password = api_key or password
if not self.password:
raise ValueError("Either password or api_key is required")
self.username = username
self.api_mode = api_mode.lower()
self.current_title = None
self.session = Session()
self.session.headers.update({"User-Agent": f"unshackle v{__version__}"})
self.api_session_id = None
if self.api_mode == "decrypt_labs":
self.session.headers.update({"decrypt-labs-api-key": self.password})
self.no_push = True
# Validate configuration based on mode
if self.api_mode == "query" and not self.username:
raise ValueError("Username is required for query mode")
elif self.api_mode not in ["query", "json", "decrypt_labs"]:
raise ValueError("api_mode must be either 'query', 'json', or 'decrypt_labs'")
def request(self, method: str, params: dict = None) -> dict:
"""Make a request to the JSON API vault."""
if self.api_mode != "json":
raise ValueError("request method is only available in json mode")
request_payload = {
"method": method,
"params": {
**(params or {}),
"session_id": self.api_session_id,
},
"token": self.password,
}
r = self.session.post(self.url, json=request_payload)
if r.status_code == 404:
return {"status": "not_found"}
if not r.ok:
raise ValueError(f"API returned HTTP Error {r.status_code}: {r.reason.title()}")
try:
res = r.json()
except json.JSONDecodeError:
if r.status_code == 404:
return {"status": "not_found"}
raise ValueError(f"API returned an invalid response: {r.text}")
if res.get("status_code") != 200:
raise ValueError(f"API returned an error: {res['status_code']} - {res['message']}")
if session_id := res.get("message", {}).get("session_id"):
self.api_session_id = session_id
return res.get("message", res)
def get_key(self, kid: Union[UUID, str], service: str) -> Optional[str]:
if isinstance(kid, UUID):
kid = kid.hex
if self.api_mode == "decrypt_labs":
try:
request_payload = {"service": service.lower(), "kid": kid}
response = self.session.post(self.url, json=request_payload)
if not response.ok:
return None
data = response.json()
if data.get("message") != "success":
return None
cached_keys = data.get("cached_keys")
if not cached_keys:
return None
if isinstance(cached_keys, str):
try:
cached_keys = json.loads(cached_keys)
except json.JSONDecodeError:
return cached_keys
if isinstance(cached_keys, dict):
if cached_keys.get("kid") == kid:
return cached_keys.get("key")
if kid in cached_keys:
return cached_keys[kid]
elif isinstance(cached_keys, list):
for entry in cached_keys:
if isinstance(entry, dict):
if entry.get("kid") == kid:
return entry.get("key")
elif isinstance(entry, str) and ":" in entry:
entry_kid, entry_key = entry.split(":", 1)
if entry_kid == kid:
return entry_key
except Exception as e:
print(f"Failed to get key from DecryptLabs ({e.__class__.__name__}: {e})")
return None
return None
elif self.api_mode == "json":
try:
params = {
"kid": kid,
"service": service.lower(),
}
response = self.request("GetKey", params)
if response.get("status") == "not_found":
return None
keys = response.get("keys", [])
for key_entry in keys:
if isinstance(key_entry, str) and ":" in key_entry:
entry_kid, entry_key = key_entry.split(":", 1)
if entry_kid == kid:
return entry_key
elif isinstance(key_entry, dict):
if key_entry.get("kid") == kid:
return key_entry.get("key")
except Exception as e:
print(f"Failed to get key ({e.__class__.__name__}: {e})")
return None
return None
else: # query mode
response = self.session.get(
self.url,
params={"service": service.lower(), "username": self.username, "password": self.password, "kid": kid},
)
data = response.json()
if data.get("status_code") != 200 or not data.get("keys"):
return None
return data["keys"][0]["key"]
def get_keys(self, service: str) -> Iterator[tuple[str, str]]:
if self.api_mode == "decrypt_labs":
return iter([])
elif self.api_mode == "json":
# JSON API doesn't support getting all keys, so return empty iterator
# This will cause the copy command to rely on the API's internal duplicate handling
return iter([])
else: # query mode
response = self.session.get(
self.url, params={"service": service.lower(), "username": self.username, "password": self.password}
)
data = response.json()
if data.get("status_code") != 200 or not data.get("keys"):
return
for key_entry in data["keys"]:
yield key_entry["kid"], key_entry["key"]
def add_key(self, service: str, kid: Union[UUID, str], key: str) -> bool:
if not key or key.count("0") == len(key):
raise ValueError("You cannot add a NULL Content Key to a Vault.")
if self.api_mode == "decrypt_labs":
return False
if isinstance(kid, UUID):
kid = kid.hex
title = getattr(self, "current_title", None)
if self.api_mode == "json":
try:
response = self.request(
"InsertKey",
{
"kid": kid,
"key": key,
"service": service.lower(),
"title": title,
},
)
if response.get("status") == "not_found":
return False
return response.get("inserted", False)
except Exception:
return False
else: # query mode
response = self.session.get(
self.url,
params={
"service": service.lower(),
"username": self.username,
"password": self.password,
"kid": kid,
"key": key,
"title": title,
},
)
data = response.json()
return data.get("status_code") == 200
def add_keys(self, service: str, kid_keys: dict[Union[UUID, str], str]) -> int:
if self.api_mode == "decrypt_labs":
return 0
for kid, key in kid_keys.items():
if not key or key.count("0") == len(key):
raise ValueError("You cannot add a NULL Content Key to a Vault.")
processed_kid_keys = {
str(kid).replace("-", "") if isinstance(kid, UUID) else kid: key for kid, key in kid_keys.items()
}
inserted_count = 0
title = getattr(self, "current_title", None)
if self.api_mode == "json":
for kid, key in processed_kid_keys.items():
try:
response = self.request(
"InsertKey",
{
"kid": kid,
"key": key,
"service": service.lower(),
"title": title,
},
)
if response.get("status") == "not_found":
continue
if response.get("inserted", False):
inserted_count += 1
except Exception:
continue
else: # query mode
for kid, key in processed_kid_keys.items():
response = self.session.get(
self.url,
params={
"service": service.lower(),
"username": self.username,
"password": self.password,
"kid": kid,
"key": key,
"title": title,
},
)
data = response.json()
if data.get("status_code") == 200 and data.get("inserted", True):
inserted_count += 1
return inserted_count
def get_services(self) -> Iterator[str]:
if self.api_mode == "decrypt_labs":
return iter([])
elif self.api_mode == "json":
try:
response = self.request("GetServices")
services = response.get("services", [])
for service in services:
yield service
except Exception:
return iter([])
else: # query mode
response = self.session.get(
self.url, params={"username": self.username, "password": self.password, "list_services": True}
)
data = response.json()
if data.get("status_code") != 200:
return
services = data.get("services", [])
for service in services:
yield service
def set_title(self, title: str):
"""
Set a title to be used for the next key insertions.
This is optional and will be sent with add_key requests if available.
"""
self.current_title = title
def insert_key_with_result(
self, service: str, kid: Union[UUID, str], key: str, title: Optional[str] = None
) -> InsertResult:
"""
Insert a key and return detailed result information.
This method provides more granular feedback than the standard add_key method.
Available in both API modes.
"""
if not key or key.count("0") == len(key):
raise ValueError("You cannot add a NULL Content Key to a Vault.")
if self.api_mode == "decrypt_labs":
return InsertResult.FAILURE
if isinstance(kid, UUID):
kid = kid.hex
if title is None:
title = getattr(self, "current_title", None)
if self.api_mode == "json":
try:
response = self.request(
"InsertKey",
{
"kid": kid,
"key": key,
"service": service.lower(),
"title": title,
},
)
if response.get("status") == "not_found":
return InsertResult.FAILURE
if response.get("inserted", False):
return InsertResult.SUCCESS
else:
return InsertResult.ALREADY_EXISTS
except Exception:
return InsertResult.FAILURE
else: # query mode
response = self.session.get(
self.url,
params={
"service": service.lower(),
"username": self.username,
"password": self.password,
"kid": kid,
"key": key,
"title": title,
},
)
try:
data = response.json()
if data.get("status_code") == 200:
if data.get("inserted", True):
return InsertResult.SUCCESS
else:
return InsertResult.ALREADY_EXISTS
else:
return InsertResult.FAILURE
except Exception:
return InsertResult.FAILURE
packages/envied/src/envied/vaults/HTTPAPI.py
+215
View File
@@ -0,0 +1,215 @@
import json
from typing import Iterator, Optional, Union
from uuid import UUID
from enum import Enum
from requests import Session
from envied.core import __version__
from envied.core.vault import Vault
class InsertResult(Enum):
FAILURE = 0
SUCCESS = 1
ALREADY_EXISTS = 2
class HTTPAPI(Vault):
"""Key Vault using a structured HTTP API with JSON payloads and token authentication."""
def __init__(self, name: str, host: str, password: str):
super().__init__(name)
self.url = host
self.password = password # This is the API token
self.current_title = None
self.session = Session()
self.session.headers.update({"User-Agent": f"Devine v{__version__}"})
self.api_session_id = None
def request(self, method: str, params: dict = None) -> dict:
"""Make a request to the HTTPAPI vault."""
request_payload = {
"method": method,
"params": {
**(params or {}),
"session_id": self.api_session_id,
},
"token": self.password,
}
r = self.session.post(self.url, json=request_payload)
if r.status_code == 404:
return {"status": "not_found"}
if not r.ok:
raise ValueError(f"API returned HTTP Error {r.status_code}: {r.reason.title()}")
try:
res = r.json()
except json.JSONDecodeError:
if r.status_code == 404:
return {"status": "not_found"}
raise ValueError(f"API returned an invalid response: {r.text}")
if res.get("status_code") != 200:
raise ValueError(f"API returned an error: {res['status_code']} - {res['message']}")
if session_id := res.get("message", {}).get("session_id"):
self.api_session_id = session_id
return res.get("message", res)
def get_key(self, kid: Union[UUID, str], service: str) -> Optional[str]:
if isinstance(kid, UUID):
kid = kid.hex
try:
title = getattr(self, "current_title", None)
response = self.request(
"GetKey",
{
"kid": kid,
"service": service.lower(),
"title": title,
},
)
if response.get("status") == "not_found":
return None
keys = response.get("keys", [])
for key_entry in keys:
if key_entry["kid"] == kid:
return key_entry["key"]
except Exception as e:
print(f"Failed to get key ({e.__class__.__name__}: {e})")
return None
return None
def get_keys(self, service: str) -> Iterator[tuple[str, str]]:
"""Get all keys for a service - this may need to be implemented based on your API."""
try:
response = self.request(
"GetKeys",
{
"service": service.lower(),
},
)
keys = response.get("keys", [])
for key_entry in keys:
yield key_entry["kid"], key_entry["key"]
except Exception:
return iter([])
def add_key(self, service: str, kid: Union[UUID, str], key: str) -> bool:
if not key or key.count("0") == len(key):
raise ValueError("You cannot add a NULL Content Key to a Vault.")
if isinstance(kid, UUID):
kid = kid.hex
title = getattr(self, "current_title", None)
try:
response = self.request(
"InsertKey",
{
"kid": kid,
"key": key,
"service": service.lower(),
"title": title,
},
)
if response.get("status") == "not_found":
return False
return response.get("inserted", False)
except Exception:
return False
def add_keys(self, service: str, kid_keys: dict[Union[UUID, str], str]) -> int:
for kid, key in kid_keys.items():
if not key or key.count("0") == len(key):
raise ValueError("You cannot add a NULL Content Key to a Vault.")
processed_kid_keys = {
str(kid).replace("-", "") if isinstance(kid, UUID) else kid: key for kid, key in kid_keys.items()
}
inserted_count = 0
title = getattr(self, "current_title", None)
for kid, key in processed_kid_keys.items():
try:
response = self.request(
"InsertKey",
{
"kid": kid,
"key": key,
"service": service.lower(),
"title": title,
},
)
if response.get("status") == "not_found":
continue
if response.get("inserted", False):
inserted_count += 1
except Exception:
continue
return inserted_count
def get_services(self) -> Iterator[str]:
"""Get available services - this may need to be implemented based on your API."""
try:
response = self.request("GetServices")
services = response.get("services", [])
for service in services:
yield service
except Exception:
return iter([])
def set_title(self, title: str):
"""
Set a title to be used for the next key insertions.
This is optional and will be sent with add_key requests if available.
"""
self.current_title = title
def insert_key_with_result(
self, service: str, kid: Union[UUID, str], key: str, title: Optional[str] = None
) -> InsertResult:
"""
Insert a key and return detailed result information.
This method provides more granular feedback than the standard add_key method.
"""
if not key or key.count("0") == len(key):
raise ValueError("You cannot add a NULL Content Key to a Vault.")
if isinstance(kid, UUID):
kid = kid.hex
if title is None:
title = getattr(self, "current_title", None)
try:
response = self.request(
"InsertKey",
{
"kid": kid,
"key": key,
"service": service.lower(),
"title": title,
},
)
if response.get("status") == "not_found":
return InsertResult.FAILURE
if response.get("inserted", False):
return InsertResult.SUCCESS
else:
return InsertResult.ALREADY_EXISTS
except Exception:
return InsertResult.FAILURE
packages/envied/src/envied/vaults/MySQL.py
+262
View File
@@ -0,0 +1,262 @@
import threading
from typing import Iterator, Optional, Union
from uuid import UUID
import pymysql
from pymysql.cursors import DictCursor
from envied.core.services import Services
from envied.core.vault import Vault
class MySQL(Vault):
"""Key Vault using a remotely-accessed mysql database connection."""
def __init__(self, name: str, host: str, database: str, username: str, no_push: bool = False, **kwargs):
"""
All extra arguments provided via **kwargs will be sent to pymysql.connect.
This can be used to provide more specific connection information.
"""
super().__init__(name, no_push)
self.slug = f"{host}:{database}:{username}"
self.conn_factory = ConnectionFactory(
dict(host=host, db=database, user=username, cursorclass=DictCursor, **kwargs)
)
self.permissions = self.get_permissions()
if not self.has_permission("SELECT"):
raise PermissionError(f"MySQL vault {self.slug} has no SELECT permission.")
def get_key(self, kid: Union[UUID, str], service: str) -> Optional[str]:
if isinstance(kid, UUID):
kid = kid.hex
service_variants = [service]
if service != service.lower():
service_variants.append(service.lower())
if service != service.upper():
service_variants.append(service.upper())
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
for service_name in service_variants:
if not self.has_table(service_name):
continue
cursor.execute(
# TODO: SQL injection risk
f"SELECT `id`, `key_` FROM `{service_name}` WHERE `kid`=%s AND `key_`!=%s",
(kid, "0" * 32),
)
cek = cursor.fetchone()
if cek:
return cek["key_"]
return None
finally:
cursor.close()
def get_keys(self, service: str) -> Iterator[tuple[str, str]]:
if not self.has_table(service):
# no table, no keys, simple
return None
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
cursor.execute(
# TODO: SQL injection risk
f"SELECT `kid`, `key_` FROM `{service}` WHERE `key_`!=%s",
("0" * 32,),
)
for row in cursor.fetchall():
yield row["kid"], row["key_"]
finally:
cursor.close()
def add_key(self, service: str, kid: Union[UUID, str], key: str) -> bool:
if not key or key.count("0") == len(key):
raise ValueError("You cannot add a NULL Content Key to a Vault.")
if not self.has_permission("INSERT", table=service):
raise PermissionError(f"MySQL vault {self.slug} has no INSERT permission.")
if not self.has_table(service):
try:
self.create_table(service)
except PermissionError:
return False
if isinstance(kid, UUID):
kid = kid.hex
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
cursor.execute(
# TODO: SQL injection risk
f"SELECT `id` FROM `{service}` WHERE `kid`=%s AND `key_`=%s",
(kid, key),
)
if cursor.fetchone():
# table already has this exact KID:KEY stored
return True
cursor.execute(
# TODO: SQL injection risk
f"INSERT INTO `{service}` (kid, key_) VALUES (%s, %s)",
(kid, key),
)
finally:
conn.commit()
cursor.close()
return True
def add_keys(self, service: str, kid_keys: dict[Union[UUID, str], str]) -> int:
for kid, key in kid_keys.items():
if not key or key.count("0") == len(key):
raise ValueError("You cannot add a NULL Content Key to a Vault.")
if not self.has_permission("INSERT", table=service):
raise PermissionError(f"MySQL vault {self.slug} has no INSERT permission.")
if not self.has_table(service):
try:
self.create_table(service)
except PermissionError:
return 0
if not isinstance(kid_keys, dict):
raise ValueError(f"The kid_keys provided is not a dictionary, {kid_keys!r}")
if not all(isinstance(kid, (str, UUID)) and isinstance(key_, str) for kid, key_ in kid_keys.items()):
raise ValueError("Expecting dict with Key of str/UUID and value of str.")
if any(isinstance(kid, UUID) for kid, key_ in kid_keys.items()):
kid_keys = {kid.hex if isinstance(kid, UUID) else kid: key_ for kid, key_ in kid_keys.items()}
if not kid_keys:
return 0
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
placeholders = ",".join(["%s"] * len(kid_keys))
cursor.execute(f"SELECT kid FROM `{service}` WHERE kid IN ({placeholders})", list(kid_keys.keys()))
existing_kids = {row["kid"] for row in cursor.fetchall()}
new_keys = {kid: key for kid, key in kid_keys.items() if kid not in existing_kids}
if not new_keys:
return 0
cursor.executemany(
f"INSERT INTO `{service}` (kid, key_) VALUES (%s, %s)",
new_keys.items(),
)
return len(new_keys)
finally:
conn.commit()
cursor.close()
def get_services(self) -> Iterator[str]:
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
cursor.execute("SHOW TABLES")
for table in cursor.fetchall():
# each entry has a key named `Tables_in_<db name>`
yield Services.get_tag(list(table.values())[0])
finally:
cursor.close()
def has_table(self, name: str) -> bool:
"""Check if the Vault has a Table with the specified name."""
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
cursor.execute(
"SELECT count(TABLE_NAME) FROM information_schema.TABLES WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s",
(conn.db, name),
)
return list(cursor.fetchone().values())[0] == 1
finally:
cursor.close()
def create_table(self, name: str):
"""Create a Table with the specified name if not yet created."""
if self.has_table(name):
return
if not self.has_permission("CREATE"):
raise PermissionError(f"MySQL vault {self.slug} has no CREATE permission.")
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
cursor.execute(
# TODO: SQL injection risk
f"""
CREATE TABLE IF NOT EXISTS {name} (
id int AUTO_INCREMENT PRIMARY KEY,
kid VARCHAR(64) NOT NULL,
key_ VARCHAR(64) NOT NULL,
UNIQUE(kid, key_)
);
"""
)
finally:
conn.commit()
cursor.close()
def get_permissions(self) -> list:
"""Get and parse Grants to a more easily usable list tuple array."""
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
cursor.execute("SHOW GRANTS")
grants = cursor.fetchall()
grants = [next(iter(x.values())) for x in grants]
grants = [tuple(x[6:].split(" TO ")[0].split(" ON ")) for x in list(grants)]
grants = [
(
list(map(str.strip, perms.replace("ALL PRIVILEGES", "*").split(","))),
location.replace("`", "").split("."),
)
for perms, location in grants
]
return grants
finally:
conn.commit()
cursor.close()
def has_permission(self, operation: str, database: Optional[str] = None, table: Optional[str] = None) -> bool:
"""Check if the current connection has a specific permission."""
grants = [x for x in self.permissions if x[0] == ["*"] or operation.upper() in x[0]]
if grants and database:
grants = [x for x in grants if x[1][0] in (database, "*")]
if grants and table:
grants = [x for x in grants if x[1][1] in (table, "*")]
return bool(grants)
class ConnectionFactory:
def __init__(self, con: dict):
self._con = con
self._store = threading.local()
def _create_connection(self) -> pymysql.Connection:
return pymysql.connect(**self._con)
def get(self) -> pymysql.Connection:
if not hasattr(self._store, "conn"):
self._store.conn = self._create_connection()
return self._store.conn
packages/envied/src/envied/vaults/SQLite.py
+216
View File
@@ -0,0 +1,216 @@
import sqlite3
import threading
from pathlib import Path
from sqlite3 import Connection
from typing import Iterator, Optional, Union
from uuid import UUID
from envied.core.services import Services
from envied.core.vault import Vault
class SQLite(Vault):
"""Key Vault using a locally-accessed sqlite DB file."""
def __init__(self, name: str, path: Union[str, Path], no_push: bool = False):
super().__init__(name, no_push)
self.path = Path(path).expanduser()
# TODO: Use a DictCursor or such to get fetches as dict?
self.conn_factory = ConnectionFactory(self.path)
def get_key(self, kid: Union[UUID, str], service: str) -> Optional[str]:
if isinstance(kid, UUID):
kid = kid.hex
conn = self.conn_factory.get()
cursor = conn.cursor()
# Try both the original service name and lowercase version to handle case sensitivity issues
service_variants = [service]
if service != service.lower():
service_variants.append(service.lower())
if service != service.upper():
service_variants.append(service.upper())
try:
for service_name in service_variants:
if not self.has_table(service_name):
continue
cursor.execute(
f"SELECT `id`, `key_` FROM `{service_name}` WHERE `kid`=? AND `key_`!=?", (kid, "0" * 32)
)
cek = cursor.fetchone()
if cek:
return cek[1]
return None
finally:
cursor.close()
def get_keys(self, service: str) -> Iterator[tuple[str, str]]:
if not self.has_table(service):
# no table, no keys, simple
return None
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
cursor.execute(f"SELECT `kid`, `key_` FROM `{service}` WHERE `key_`!=?", ("0" * 32,))
for kid, key_ in cursor.fetchall():
yield kid, key_
finally:
cursor.close()
def add_key(self, service: str, kid: Union[UUID, str], key: str) -> bool:
if not key or key.count("0") == len(key):
raise ValueError("You cannot add a NULL Content Key to a Vault.")
if not self.has_table(service):
self.create_table(service)
if isinstance(kid, UUID):
kid = kid.hex
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
cursor.execute(
# TODO: SQL injection risk
f"SELECT `id` FROM `{service}` WHERE `kid`=? AND `key_`=?",
(kid, key),
)
if cursor.fetchone():
# table already has this exact KID:KEY stored
return True
cursor.execute(
# TODO: SQL injection risk
f"INSERT INTO `{service}` (kid, key_) VALUES (?, ?)",
(kid, key),
)
finally:
conn.commit()
cursor.close()
return True
def add_keys(self, service: str, kid_keys: dict[Union[UUID, str], str]) -> int:
for kid, key in kid_keys.items():
if not key or key.count("0") == len(key):
raise ValueError("You cannot add a NULL Content Key to a Vault.")
if not self.has_table(service):
self.create_table(service)
if not isinstance(kid_keys, dict):
raise ValueError(f"The kid_keys provided is not a dictionary, {kid_keys!r}")
if not all(isinstance(kid, (str, UUID)) and isinstance(key_, str) for kid, key_ in kid_keys.items()):
raise ValueError("Expecting dict with Key of str/UUID and value of str.")
if any(isinstance(kid, UUID) for kid, key_ in kid_keys.items()):
kid_keys = {kid.hex if isinstance(kid, UUID) else kid: key_ for kid, key_ in kid_keys.items()}
if not kid_keys:
return 0
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
# Query existing KIDs in batches to avoid SQLite variable limit
# Try larger batch first (newer SQLite supports 32766), fall back to 500 if needed
existing_kids: set[str] = set()
kid_list = list(kid_keys.keys())
batch_size = 32000
i = 0
while i < len(kid_list):
batch = kid_list[i : i + batch_size]
placeholders = ",".join(["?"] * len(batch))
try:
cursor.execute(f"SELECT kid FROM `{service}` WHERE kid IN ({placeholders})", batch)
existing_kids.update(row[0] for row in cursor.fetchall())
i += batch_size
except sqlite3.OperationalError as e:
if "too many SQL variables" in str(e) and batch_size > 500:
batch_size = 500
continue
raise
new_keys = {kid: key for kid, key in kid_keys.items() if kid not in existing_kids}
if not new_keys:
return 0
cursor.executemany(
f"INSERT INTO `{service}` (kid, key_) VALUES (?, ?)",
new_keys.items(),
)
return len(new_keys)
finally:
conn.commit()
cursor.close()
def get_services(self) -> Iterator[str]:
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
for (name,) in cursor.fetchall():
if name != "sqlite_sequence":
yield Services.get_tag(name)
finally:
cursor.close()
def has_table(self, name: str) -> bool:
"""Check if the Vault has a Table with the specified name."""
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
cursor.execute("SELECT count(name) FROM sqlite_master WHERE type='table' AND name=?", (name,))
return cursor.fetchone()[0] == 1
finally:
cursor.close()
def create_table(self, name: str):
"""Create a Table with the specified name if not yet created."""
if self.has_table(name):
return
conn = self.conn_factory.get()
cursor = conn.cursor()
try:
cursor.execute(
# TODO: SQL injection risk
f"""
CREATE TABLE IF NOT EXISTS {name} (
"id" INTEGER NOT NULL UNIQUE,
"kid" TEXT NOT NULL COLLATE NOCASE,
"key_" TEXT NOT NULL COLLATE NOCASE,
PRIMARY KEY("id" AUTOINCREMENT),
UNIQUE("kid", "key_")
);
"""
)
finally:
conn.commit()
cursor.close()
class ConnectionFactory:
def __init__(self, path: Union[str, Path]):
self._path = path
self._store = threading.local()
def _create_connection(self) -> Connection:
return sqlite3.connect(self._path)
def get(self) -> Connection:
if not hasattr(self._store, "conn"):
self._store.conn = self._create_connection()
return self._store.conn
packages/envied/src/envied/vaults/__init__.py
View File