(cls, database_info: DatabaseConnection)
| 142 | |
| 143 | @classmethod |
| 144 | def from_uri_ssh(cls, database_info: DatabaseConnection): |
| 145 | file_path = database_info.path_to_credentials_file |
| 146 | if file_path.lower().startswith("s3"): |
| 147 | s3 = S3() |
| 148 | file_path = s3.download(file_path) |
| 149 | |
| 150 | fernet_encrypt = FernetEncrypt() |
| 151 | db_uri = unquote(fernet_encrypt.decrypt(database_info.connection_uri)) |
| 152 | db_uri_obj = cls.extract_parameters(db_uri) |
| 153 | ssh = database_info.ssh_settings |
| 154 | |
| 155 | server = SSHTunnelForwarder( |
| 156 | (ssh.host, 22 if not ssh.port else int(ssh.port)), |
| 157 | ssh_username=ssh.username, |
| 158 | ssh_password=fernet_encrypt.decrypt(ssh.password), |
| 159 | ssh_pkey=file_path, |
| 160 | ssh_private_key_password=fernet_encrypt.decrypt(ssh.private_key_password), |
| 161 | remote_bind_address=( |
| 162 | db_uri_obj["host"], |
| 163 | 5432 if not db_uri_obj["port"] else int(db_uri_obj["port"]), |
| 164 | ), |
| 165 | ) |
| 166 | server.stop(force=True) |
| 167 | server.start() |
| 168 | local_port = str(server.local_bind_port) |
| 169 | local_host = str(server.local_bind_host) |
| 170 | |
| 171 | return cls.from_uri( |
| 172 | f"{db_uri_obj['driver']}://{db_uri_obj['user']}:{db_uri_obj['password']}@{local_host}:{local_port}/{db_uri_obj['db']}" |
| 173 | ) |
| 174 | |
| 175 | @classmethod |
| 176 | def parser_to_filter_commands(cls, command: str) -> str: |
no test coverage detected