Skip to content

ark_sia_ssh_ca_service

ArkSIASSHCAService

Bases: ArkService

Source code in ark_sdk_python/services/sia/ssh_ca/ark_sia_ssh_ca_service.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class ArkSIASSHCAService(ArkService):
    def __init__(self, isp_auth: ArkISPAuth) -> None:
        super().__init__(isp_auth)
        self.__isp_auth = isp_auth
        self.__client: ArkISPServiceClient = ArkISPServiceClient.from_isp_auth(
            isp_auth=self.__isp_auth,
            service_name='dpa',
            refresh_connection_callback=self.__refresh_sia_auth,
        )

    def __refresh_sia_auth(self, client: ArkISPServiceClient) -> None:
        ArkISPServiceClient.refresh_client(client, self.__isp_auth)

    def generate_new_ca(self) -> None:
        """
        Generate new SSH CA key version

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info('Generate new CA key version')
        resp: Response = self.__client.post(GENERATE_NEW_CA_KEY_API)
        if resp.status_code != HTTPStatus.CREATED:
            raise ArkServiceException(f'Failed to generate new CA key [{resp.text}] - [{resp.status_code}]')

    def deactivate_previous_ca(self) -> None:
        """
        Deactivate previous SSH CA key version

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info('Deactivate previous CA key version')
        resp: Response = self.__client.post(DEACTIVATE_PREVIOUS_CA_KEY_API)
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to deactivate previous CA key [{resp.text}] - [{resp.status_code}]')

    def reactivate_previous_ca(self) -> None:
        """
        Reactivate previous SSH CA key version

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info('Reactivate previous CA key version')
        resp: Response = self.__client.post(REACTIVATE_PREVIOUS_CA_KEY_API)
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to reactivate previous CA key [{resp.text}] - [{resp.status_code}]')

    def public_key(self, get_public_key: ArkSIAGetSSHPublicKey) -> str:
        """
        Retrieves the public key used for SIA SSH connections trust with customer env

        Args:
            get_public_key (ArkSIAGetSSHPublicKey): _description_

        Raises:
            ArkNotSupportedException: _description_
            ArkServiceException: _description_

        Returns:
            str
        """
        self._logger.info('Getting public key')
        resp: Response = self.__client.get(PUBLIC_KEYS_API)
        if resp.status_code == HTTPStatus.OK:
            if get_public_key.output_file:
                os.makedirs(os.path.dirname(get_public_key.output_file), exist_ok=True)
                with open(get_public_key.output_file, 'w', encoding='utf-8') as f:
                    f.write(resp.text)
            return resp.text
        raise ArkServiceException(f'Failed to get public key [{resp.text}] - [{resp.status_code}]')

    def public_key_script(self, get_public_key: ArkSIAGetSSHPublicKey) -> str:
        """
        Retrieves the public key script used for SIA SSH connections trust with customer env
        The script can be run to install the public key in needed ssh configuration files

        Args:
            get_public_key (ArkSIAGetSSHPublicKey): _description_

        Raises:
            ArkNotSupportedException: _description_
            ArkServiceException: _description_

        Returns:
            str
        """
        self._logger.info('Getting public key script')
        resp: Response = self.__client.get(
            PUBLIC_KEYS_SCRIPT_API,
        )
        if resp.status_code == HTTPStatus.OK:
            if get_public_key.output_file:
                os.makedirs(os.path.dirname(get_public_key.output_file), exist_ok=True)
                with open(get_public_key.output_file, 'w', encoding='utf-8') as f:
                    f.write(resp.text)
            return resp.text
        raise ArkServiceException(f'Failed to get public key script [{resp.text}] - [{resp.status_code}]')

    @staticmethod
    @overrides
    def service_config() -> ArkServiceConfig:
        return SERVICE_CONFIG

deactivate_previous_ca()

Deactivate previous SSH CA key version

Raises:

Type Description
ArkServiceException

description
Source code in ark_sdk_python/services/sia/ssh_ca/ark_sia_ssh_ca_service.py 
53
54
55
56
57
58
59
60
61
62
63
def deactivate_previous_ca(self) -> None:
    """
    Deactivate previous SSH CA key version

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info('Deactivate previous CA key version')
    resp: Response = self.__client.post(DEACTIVATE_PREVIOUS_CA_KEY_API)
    if resp.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to deactivate previous CA key [{resp.text}] - [{resp.status_code}]')

generate_new_ca()

Generate new SSH CA key version

Raises:

Type Description
ArkServiceException

description
Source code in ark_sdk_python/services/sia/ssh_ca/ark_sia_ssh_ca_service.py 
41
42
43
44
45
46
47
48
49
50
51
def generate_new_ca(self) -> None:
    """
    Generate new SSH CA key version

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info('Generate new CA key version')
    resp: Response = self.__client.post(GENERATE_NEW_CA_KEY_API)
    if resp.status_code != HTTPStatus.CREATED:
        raise ArkServiceException(f'Failed to generate new CA key [{resp.text}] - [{resp.status_code}]')

public_key(get_public_key)

Retrieves the public key used for SIA SSH connections trust with customer env

Parameters:

Name Type Description Default
get_public_key ArkSIAGetSSHPublicKey

description

 required

Raises:

Type Description
ArkNotSupportedException

description
ArkServiceException

description

Returns:

Type Description
str

str
Source code in ark_sdk_python/services/sia/ssh_ca/ark_sia_ssh_ca_service.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def public_key(self, get_public_key: ArkSIAGetSSHPublicKey) -> str:
    """
    Retrieves the public key used for SIA SSH connections trust with customer env

    Args:
        get_public_key (ArkSIAGetSSHPublicKey): _description_

    Raises:
        ArkNotSupportedException: _description_
        ArkServiceException: _description_

    Returns:
        str
    """
    self._logger.info('Getting public key')
    resp: Response = self.__client.get(PUBLIC_KEYS_API)
    if resp.status_code == HTTPStatus.OK:
        if get_public_key.output_file:
            os.makedirs(os.path.dirname(get_public_key.output_file), exist_ok=True)
            with open(get_public_key.output_file, 'w', encoding='utf-8') as f:
                f.write(resp.text)
        return resp.text
    raise ArkServiceException(f'Failed to get public key [{resp.text}] - [{resp.status_code}]')

public_key_script(get_public_key)

Retrieves the public key script used for SIA SSH connections trust with customer env The script can be run to install the public key in needed ssh configuration files

Parameters:

Name Type Description Default
get_public_key ArkSIAGetSSHPublicKey

description

 required

Raises:

Type Description
ArkNotSupportedException

description
ArkServiceException

description

Returns:

Type Description
str

str
Source code in ark_sdk_python/services/sia/ssh_ca/ark_sia_ssh_ca_service.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def public_key_script(self, get_public_key: ArkSIAGetSSHPublicKey) -> str:
    """
    Retrieves the public key script used for SIA SSH connections trust with customer env
    The script can be run to install the public key in needed ssh configuration files

    Args:
        get_public_key (ArkSIAGetSSHPublicKey): _description_

    Raises:
        ArkNotSupportedException: _description_
        ArkServiceException: _description_

    Returns:
        str
    """
    self._logger.info('Getting public key script')
    resp: Response = self.__client.get(
        PUBLIC_KEYS_SCRIPT_API,
    )
    if resp.status_code == HTTPStatus.OK:
        if get_public_key.output_file:
            os.makedirs(os.path.dirname(get_public_key.output_file), exist_ok=True)
            with open(get_public_key.output_file, 'w', encoding='utf-8') as f:
                f.write(resp.text)
        return resp.text
    raise ArkServiceException(f'Failed to get public key script [{resp.text}] - [{resp.status_code}]')

reactivate_previous_ca()

Reactivate previous SSH CA key version

Raises:

Type Description
ArkServiceException

description
Source code in ark_sdk_python/services/sia/ssh_ca/ark_sia_ssh_ca_service.py 
65
66
67
68
69
70
71
72
73
74
75
def reactivate_previous_ca(self) -> None:
    """
    Reactivate previous SSH CA key version

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info('Reactivate previous CA key version')
    resp: Response = self.__client.post(REACTIVATE_PREVIOUS_CA_KEY_API)
    if resp.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to reactivate previous CA key [{resp.text}] - [{resp.status_code}]')