vendor_connectors.secrets
Secrets Connector - Enterprise-grade secret synchronization.
This connector provides Python bindings for secretssync, enabling enterprise-grade secret synchronization from HashiCorp Vault to AWS Secrets Manager with two-phase architecture, inheritance, versioning, and CI/CD integration.
The connector can operate in two modes:
- Native mode: Uses gopy-generated Python bindings for maximum performance
- CLI mode: Falls back to subprocess calls if bindings aren’t available
Example usage: from vendor_connectors.secrets import SecretsConnector
# Initialize connectorconnector = SecretsConnector()
# Validate a configurationis_valid, message = connector.validate_config("pipeline.yaml")
# Run a dry-run to see what would changeresult = connector.dry_run("pipeline.yaml")print(f"Would sync {result.secrets_processed} secrets")
# Execute the full pipelineresult = connector.run_pipeline("pipeline.yaml")if result.success: print(f"Synced {result.secrets_added} secrets")Submodules
Section titled “Submodules”Package Contents
Section titled “Package Contents”Classes
Section titled “Classes”SyncOperation | Pipeline operation types. |
|---|---|
OutputFormat | Output format for diff display. |
SyncOptions | Options for pipeline execution. |
SyncResult | Result of a sync operation. |
ConfigInfo | Information about a pipeline configuration. |
SecretsConnector | Enterprise-grade secret synchronization connector. |
_NATIVE_AVAILABLE |
|---|
vendor_connectors.secrets._NATIVE_AVAILABLE
Section titled “vendor_connectors.secrets._NATIVE_AVAILABLE”False
class vendor_connectors.secrets.SyncOperation
Section titled “class vendor_connectors.secrets.SyncOperation”Pipeline operation types.
Initialization
Section titled “Initialization”Initialize self. See help(type(self)) for accurate signature.
‘merge’
‘sync’
PIPELINE
Section titled “PIPELINE”‘pipeline’
class vendor_connectors.secrets.OutputFormat
Section titled “class vendor_connectors.secrets.OutputFormat”Output format for diff display.
Initialization
Section titled “Initialization”Initialize self. See help(type(self)) for accurate signature.
‘human’
‘json’
GITHUB
Section titled “GITHUB”‘github’
COMPACT
Section titled “COMPACT”‘compact’
SIDE_BY_SIDE
Section titled “SIDE_BY_SIDE”‘side-by-side’
class vendor_connectors.secrets.SyncOptions
Section titled “class vendor_connectors.secrets.SyncOptions”Options for pipeline execution.
dry_run : bool
Section titled “dry_run : bool”False
operation : vendor_connectors.secrets.SyncOperation
Section titled “operation : vendor_connectors.secrets.SyncOperation”None
targets : list[str]
Section titled “targets : list[str]”‘field(…)’
continue_on_error : bool
Section titled “continue_on_error : bool”False
parallelism : int
Section titled “parallelism : int”4
compute_diff : bool
Section titled “compute_diff : bool”False
output_format : vendor_connectors.secrets.OutputFormat
Section titled “output_format : vendor_connectors.secrets.OutputFormat”None
class vendor_connectors.secrets.SyncResult
Section titled “class vendor_connectors.secrets.SyncResult”Result of a sync operation.
success : bool
Section titled “success : bool”False
target_count : int
Section titled “target_count : int”0
secrets_processed : int
Section titled “secrets_processed : int”0
secrets_added : int
Section titled “secrets_added : int”0
secrets_modified : int
Section titled “secrets_modified : int”0
secrets_removed : int
Section titled “secrets_removed : int”0
secrets_unchanged : int
Section titled “secrets_unchanged : int”0
duration_ms : int
Section titled “duration_ms : int”0
error_message : str =
Section titled “error_message : str = ”results_json : str =
Section titled “results_json : str = ”diff_output : str =
Section titled “diff_output : str = ”classmethod from_native(native_result) → vendor_connectors.secrets.SyncResult
Section titled “classmethod from_native(native_result) → vendor_connectors.secrets.SyncResult”Create from native gopy result.
classmethod from_cli_output(output: dict) → vendor_connectors.secrets.SyncResult
Section titled “classmethod from_cli_output(output: dict) → vendor_connectors.secrets.SyncResult”Create from CLI JSON output.
class vendor_connectors.secrets.ConfigInfo
Section titled “class vendor_connectors.secrets.ConfigInfo”Information about a pipeline configuration.
False
error_message : str =
Section titled “error_message : str = ”source_count : int
Section titled “source_count : int”0
target_count : int
Section titled “target_count : int”0
sources : list[str]
Section titled “sources : list[str]”‘field(…)’
targets : list[str]
Section titled “targets : list[str]”‘field(…)’
has_merge_store : bool
Section titled “has_merge_store : bool”False
vault_address : str =
Section titled “vault_address : str = ”aws_region : str =
Section titled “aws_region : str = ”classmethod from_native(native_info) → vendor_connectors.secrets.ConfigInfo
Section titled “classmethod from_native(native_info) → vendor_connectors.secrets.ConfigInfo”Create from native gopy result.
class vendor_connectors.secrets.SecretsConnector(cli_path: str | None = None, prefer_native: bool = True, logger: lifecyclelogging.Logging | None = None, **kwargs)
Section titled “class vendor_connectors.secrets.SecretsConnector(cli_path: str | None = None, prefer_native: bool = True, logger: lifecyclelogging.Logging | None = None, **kwargs)”Bases: vendor_connectors.base.VendorConnectorBase
Enterprise-grade secret synchronization connector.
This connector wraps the secretssync Go library, providing Python bindings for enterprise-grade secret synchronization between HashiCorp Vault and AWS Secrets Manager.
Features:
- Two-phase pipeline architecture (merge → sync)
- Secret inheritance and deep merging
- AWS Organizations discovery
- Dry-run with visual diff output
- CI/CD integration with exit codes
The connector operates in two modes:
- Native mode: Uses gopy-generated bindings (faster)
- CLI mode: Falls back to subprocess if bindings unavailable
Initialization
Section titled “Initialization”Initialize the secrets connector.
Args: cli_path: Path to secretsync CLI binary (for CLI mode) prefer_native: Prefer native bindings over CLI logger: Logger instance **kwargs: Passed to VendorConnectorBase
_find_cli() → str | None
Section titled “_find_cli() → str | None”Find the secretsync CLI binary.
property native_available : bool
Section titled “property native_available : bool”Check if native bindings are available.
property cli_available : bool
Section titled “property cli_available : bool”Check if CLI is available.
validate_config(config_path: str) → tuple[bool, str]
Section titled “validate_config(config_path: str) → tuple[bool, str]”Validate a pipeline configuration file.
Args: config_path: Path to YAML configuration file
Returns: Tuple of (is_valid, message)
_cli_validate_config(config_path: str) → tuple[bool, str]
Section titled “_cli_validate_config(config_path: str) → tuple[bool, str]”Validate config via CLI.
get_config_info(config_path: str) → vendor_connectors.secrets.ConfigInfo
Section titled “get_config_info(config_path: str) → vendor_connectors.secrets.ConfigInfo”Get detailed information about a configuration.
Args: config_path: Path to YAML configuration file
Returns: ConfigInfo with configuration details
_cli_get_config_info(config_path: str) → vendor_connectors.secrets.ConfigInfo
Section titled “_cli_get_config_info(config_path: str) → vendor_connectors.secrets.ConfigInfo”Get config info via CLI.
run_pipeline(config_path: str, options: vendor_connectors.secrets.SyncOptions | None = None) → vendor_connectors.secrets.SyncResult
Section titled “run_pipeline(config_path: str, options: vendor_connectors.secrets.SyncOptions | None = None) → vendor_connectors.secrets.SyncResult”Execute the secrets synchronization pipeline.
Args: config_path: Path to YAML configuration file options: Execution options (defaults to full pipeline)
Returns: SyncResult with operation details
_native_run_pipeline(config_path: str, options: vendor_connectors.secrets.SyncOptions) → vendor_connectors.secrets.SyncResult
Section titled “_native_run_pipeline(config_path: str, options: vendor_connectors.secrets.SyncOptions) → vendor_connectors.secrets.SyncResult”Run pipeline via native bindings.
_cli_run_pipeline(config_path: str, options: vendor_connectors.secrets.SyncOptions) → vendor_connectors.secrets.SyncResult
Section titled “_cli_run_pipeline(config_path: str, options: vendor_connectors.secrets.SyncOptions) → vendor_connectors.secrets.SyncResult”Run pipeline via CLI.
dry_run(config_path: str) → vendor_connectors.secrets.SyncResult
Section titled “dry_run(config_path: str) → vendor_connectors.secrets.SyncResult”Perform a dry run of the pipeline.
Args: config_path: Path to YAML configuration file
Returns: SyncResult with what would be changed
merge(config_path: str, dry_run: bool = False) → vendor_connectors.secrets.SyncResult
Section titled “merge(config_path: str, dry_run: bool = False) → vendor_connectors.secrets.SyncResult”Run only the merge phase of the pipeline.
Args: config_path: Path to YAML configuration file dry_run: If True, don’t make actual changes
Returns: SyncResult with merge operation details
sync(config_path: str, dry_run: bool = False) → vendor_connectors.secrets.SyncResult
Section titled “sync(config_path: str, dry_run: bool = False) → vendor_connectors.secrets.SyncResult”Run only the sync phase of the pipeline.
Args: config_path: Path to YAML configuration file dry_run: If True, don’t make actual changes
Returns: SyncResult with sync operation details
get_targets(config_path: str) → tuple[list[str], str]
Section titled “get_targets(config_path: str) → tuple[list[str], str]”Get the list of targets from a configuration.
Args: config_path: Path to YAML configuration file
Returns: Tuple of (targets, error_message)
get_sources(config_path: str) → tuple[list[str], str]
Section titled “get_sources(config_path: str) → tuple[list[str], str]”Get the list of sources from a configuration.
Args: config_path: Path to YAML configuration file
Returns: Tuple of (sources, error_message)