Source code for hvac.api.system_backend.raft

#!/usr/bin/env python
"""Raft methods module."""
from hvac.api.system_backend.system_backend_mixin import SystemBackendMixin
from hvac import utils, adapters


[docs]class Raft(SystemBackendMixin): """Raft cluster-related system backend methods. When using Shamir seal, as soon as the Vault server is brought up, this API should be invoked instead of sys/init. This API completes in 2 phases. Once this is invoked, the joining node will receive a challenge from the Raft's leader node. This challenge can be answered by the joining node only after a successful unseal. Hence, the joining node should be unsealed using the unseal keys of the Raft's leader node. Reference: https://www.vaultproject.io/api-docs/system/storage/raft """
[docs] def join_raft_cluster( self, leader_api_addr, retry=False, leader_ca_cert=None, leader_client_cert=None, leader_client_key=None, ): """Join a new server node to the Raft cluster. When using Shamir seal, as soon as the Vault server is brought up, this API should be invoked instead of sys/init. This API completes in 2 phases. Once this is invoked, the joining node will receive a challenge from the Raft's leader node. This challenge can be answered by the joining node only after a successful unseal. Hence, the joining node should be unsealed using the unseal keys of the Raft's leader node. Supported methods: POST: /sys/storage/raft/join. :param leader_api_addr: Address of the leader node in the Raft cluster to which this node is trying to join. :type leader_api_addr: str | unicode :param retry: Retry joining the Raft cluster in case of failures. :type retry: bool :param leader_ca_cert: CA certificate used to communicate with Raft's leader node. :type leader_ca_cert: str | unicode :param leader_client_cert: Client certificate used to communicate with Raft's leader node. :type leader_client_cert: str | unicode :param leader_client_key: Client key used to communicate with Raft's leader node. :type leader_client_key: str | unicode :return: The response of the join_raft_cluster request. :rtype: requests.Response """ params = utils.remove_nones( { "leader_api_addr": leader_api_addr, "retry": retry, "leader_ca_cert": leader_ca_cert, "leader_client_cert": leader_client_cert, "leader_client_key": leader_client_key, } ) api_path = "/v1/sys/storage/raft/join" return self._adapter.post( url=api_path, json=params, )
[docs] def read_raft_config(self): """Read the details of all the nodes in the raft cluster. Supported methods: GET: /sys/storage/raft/configuration. :return: The response of the read_raft_config request. :rtype: requests.Response """ api_path = "/v1/sys/storage/raft/configuration" return self._adapter.get( url=api_path, )
[docs] def remove_raft_node(self, server_id): """Remove a node from the raft cluster. Supported methods: POST: /sys/storage/raft/remove-peer. :param server_id: The ID of the node to remove. :type server_id: str :return: The response of the remove_raft_node request. :rtype: requests.Response """ params = { "server_id": server_id, } api_path = "/v1/sys/storage/raft/remove-peer" return self._adapter.post( url=api_path, json=params, )
[docs] def take_raft_snapshot(self): """Returns a snapshot of the current state of the raft cluster. The snapshot is returned as binary data and should be redirected to a file. This endpoint will ignore your chosen adapter and always uses a RawAdapter. Supported methods: GET: /sys/storage/raft/snapshot. :return: The response of the snapshot request. :rtype: requests.Response """ api_path = "/v1/sys/storage/raft/snapshot" raw_adapter = adapters.RawAdapter.from_adapter(self._adapter) return raw_adapter.get( url=api_path, stream=True, )
[docs] def restore_raft_snapshot(self, snapshot): """Install the provided snapshot, returning the cluster to the state defined in it. Supported methods: POST: /sys/storage/raft/snapshot. :param snapshot: Previously created raft snapshot / binary data. :type snapshot: bytes :return: The response of the restore_raft_snapshot request. :rtype: requests.Response """ api_path = "/v1/sys/storage/raft/snapshot" return self._adapter.post( url=api_path, data=snapshot, )
[docs] def force_restore_raft_snapshot(self, snapshot): """Installs the provided snapshot, returning the cluster to the state defined in it. This is same as writing to /sys/storage/raft/snapshot except that this bypasses checks ensuring the Autounseal or shamir keys are consistent with the snapshot data. Supported methods: POST: /sys/storage/raft/snapshot-force. :param snapshot: Previously created raft snapshot / binary data. :type snapshot: bytes :return: The response of the force_restore_raft_snapshot request. :rtype: requests.Response """ api_path = "/v1/sys/storage/raft/snapshot-force" return self._adapter.post( url=api_path, data=snapshot, )
[docs] def read_raft_auto_snapshot_status(self, name): """Read the status of the raft auto snapshot. Supported methods: GET: /sys/storage/raft/snapshot-auto/status/:name. Produces: 200 application/json :param name: The name of the snapshot configuration. :type name: str :return: The response of the read_raft_auto_snapshot_status request. :rtype: requests.Response """ api_path = f"/v1/sys/storage/raft/snapshot-auto/status/{name}" return self._adapter.get( url=api_path, )
[docs] def read_raft_auto_snapshot_config(self, name): """Read the configuration of the raft auto snapshot. Supported methods: GET: /sys/storage/raft/snapshot-auto/config/:name. Produces: 200 application/json :param name: The name of the snapshot configuration. :type name: str :return: The response of the read_raft_auto_snapshot_config request. :rtype: requests.Response """ api_path = f"/v1/sys/storage/raft/snapshot-auto/config/{name}" return self._adapter.get( url=api_path, )
[docs] def list_raft_auto_snapshot_configs(self): """List the configurations of the raft auto snapshot. Supported methods: LIST: /sys/storage/raft/snapshot-auto/config. Produces: 200 application/json :return: The response of the list_raft_auto_snapshot_configs request. :rtype: requests.Response """ api_path = "/v1/sys/storage/raft/snapshot-auto/config" return self._adapter.list( url=api_path, )
[docs] def create_or_update_raft_auto_snapshot_config( self, name, interval, storage_type, retain=1, **kwargs ): """Create or update the configuration of the raft auto snapshot. Supported methods: POST: /sys/storage/raft/snapshot-auto/config/:name. Produces: 204 application/json :param name: The name of the snapshot configuration. :type name: str :param interval: The interval at which snapshots should be taken. :type interval: str :param storage_type: The type of storage to use for the snapshot. :type storage_type: str :param retain: The number of snapshots to retain. Default is 1 :type retain: int :param kwargs: Additional parameters to send in the request. Should be params specific to the storage type. :type kwargs: dict :return: The response of the create_or_update_raft_auto_snapshot_config request. :rtype: requests.Response """ params = utils.remove_nones( { "interval": interval, "storage_type": storage_type, "retain": retain, **kwargs, } ) api_path = f"/v1/sys/storage/raft/snapshot-auto/config/{name}" return self._adapter.post( url=api_path, json=params, )
[docs] def delete_raft_auto_snapshot_config(self, name): """Delete the configuration of the raft auto snapshot. Supported methods: DELETE: /sys/storage/raft/snapshot-auto/config/:name. Produces: 204 application/json :param name: The name of the snapshot configuration. :type name: str :return: The response of the delete_raft_auto_snapshot_config request. :rtype: requests.Response """ api_path = f"/v1/sys/storage/raft/snapshot-auto/config/{name}" return self._adapter.delete( url=api_path, )