db_sync_tool.remote.client.load_ssh_client_target() - Code Metrics - jackd248/db-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

load_ssh_client_target() A
last analyzed 2025-04-09 15:26 UTC

↳ Parent: db_sync_tool.remote.client

Complexity

Conditions

Size

Total Lines	8
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	8
rs	10
c	0
b	0
f	0
cc	1
nop	0

#!/usr/bin/env python3
# -*- coding: future_fstrings -*-

"""
Client script
"""

import sys
import paramiko
from db_sync_tool.utility import mode, system, helper, output

ssh_client_origin = None

ssh_client_target = None

additional_ssh_clients = []

default_timeout = 600



def load_ssh_client_origin():
    """
    Loading the origin ssh client
    :return:
    """
    global ssh_client_origin

    ssh_client_origin = load_ssh_client(mode.Client.ORIGIN)
    helper.run_script(mode.Client.ORIGIN, 'before')


def load_ssh_client_target():
    """
    Loading the target ssh client
    :return:
    """
    global ssh_client_target

    ssh_client_target = load_ssh_client(mode.Client.TARGET)
    helper.run_script(mode.Client.TARGET, 'before')


def load_ssh_client(ssh):
    """
    Initializing the given ssh client
    :param ssh: String
    :return:
    """
    _host_name = helper.get_ssh_host_name(ssh, True)
    _ssh_client = paramiko.SSHClient()
    _ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    _ssh_port = system.config[ssh]['port'] if 'port' in system.config[ssh] else 22
    _ssh_key = None
    _ssh_password = None

    # Check authentication
    if 'ssh_key' in system.config[ssh]:
        _authentication_method = f'{output.CliFormat.BLACK} - ' \
                                 f'(authentication: key){output.CliFormat.ENDC}'
        _ssh_key = system.config[ssh]['ssh_key']
    elif 'password' in system.config[ssh]:
        _authentication_method = f'{output.CliFormat.BLACK} - ' \
                                 f'authentication: password){output.CliFormat.ENDC}'
        _ssh_password = system.config[ssh]['password']
    elif 'ssh_agent' in system.config:
        _authentication_method = f'{output.CliFormat.BLACK} - ' \
                                 f'(authentication: key){output.CliFormat.ENDC}'
    else:
        sys.exit(
            output.message(
                output.Subject.ERROR,
                'Missing SSH authentication. Neither ssh key nor ssh password given.',
                False
            )
        )

    # Try to connect to remote client via paramiko
    try:
        _ssh_client.connect(hostname=system.config[ssh]['host'],
                            username=system.config[ssh]['user'],
                            key_filename=_ssh_key,
                            password=_ssh_password,
                            port=_ssh_port,
                            compress=True,
                            timeout=default_timeout,
                            sock=get_jump_host_channel(ssh))
        #
        # Workaround for long-lasting requests
        # https://stackoverflow.com/questions/50009688/python-paramiko-ssh-session-not-active-after-being-idle-for-many-hours
        #
        _ssh_client.get_transport().set_keepalive(60)

    except paramiko.ssh_exception.AuthenticationException:
        sys.exit(
            output.message(
                output.Subject.ERROR,
                f'SSH authentication for {_host_name} failed',
                False
            )
        )

    output.message(
        output.host_to_subject(ssh),
        f'Initialize remote SSH connection {_host_name}{_authentication_method}',
        True
    )

    return _ssh_client


def close_ssh_clients():
    """
    Closing ssh client sessions
    :return:
    """
    helper.run_script(mode.Client.ORIGIN, 'after')
    if not ssh_client_origin is None:
        ssh_client_origin.close()

    helper.run_script(mode.Client.TARGET, 'after')
    if not ssh_client_target is None:
        ssh_client_target.close()

    for additional_ssh_client in additional_ssh_clients:
        additional_ssh_client.close()

    helper.run_script(script='after')


def get_jump_host_channel(client):
    """
    Provide an optional transport channel for a SSH jump host client
    https://gist.github.com/tintoy/443c42ea3865680cd624039c4bb46219
    :param client:
    :return:
    """
    _jump_host_channel = None
    if 'jump_host' in system.config[client]:
        # prepare jump host config
        _jump_host_client = paramiko.SSHClient()
        _jump_host_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        _jump_host_host = system.config[client]['jump_host']['host']
        _jump_host_user = system.config[client]['jump_host']['user'] if 'user' in system.config[client]['jump_host'] else system.config[client]['user']


        if 'ssh_key' in system.config[client]['jump_host']:
            _jump_host_ssh_key = system.config[client]['jump_host']['ssh_key']
        elif 'ssh_key' in system.config[client]:
            _jump_host_ssh_key = system.config[client]['ssh_key']
        else:
            _jump_host_ssh_key = None

        if 'port' in system.config[client]['jump_host']:
            _jump_host_port = system.config[client]['jump_host']['port']
        elif 'port' in system.config[client]:
            _jump_host_port = system.config[client]['port']
        else:
            _jump_host_port = 22

        # connect to the jump host
        _jump_host_client.connect(
            hostname=_jump_host_host,
            username=_jump_host_user,
            key_filename=_jump_host_ssh_key,
            password=system.config[client]['jump_host']['password'] if 'password' in system.config[client]['jump_host'] else None,

            port=_jump_host_port,
            compress=True,
            timeout=default_timeout
        )

        global additional_ssh_clients

        additional_ssh_clients.append(_jump_host_client)

        # open the necessary channel
        _jump_host_transport = _jump_host_client.get_transport()
        _jump_host_channel = _jump_host_transport.open_channel(
            'direct-tcpip',
            dest_addr=(system.config[client]['host'], 22),
            src_addr=(system.config[client]['jump_host']['private'] if 'private' in system.config[client]['jump_host'] else system.config[client]['jump_host']['host'], 22)

        )

        # print information
        _destination_client = helper.get_ssh_host_name(client, minimal=True)
        _jump_host_name = system.config[client]['jump_host']['name'] if 'name' in system.config[client]['jump_host'] else _jump_host_host

        output.message(
            output.host_to_subject(client),
            f'Initialize remote SSH jump host {output.CliFormat.BLACK}local ➔ {output.CliFormat.BOLD}{_jump_host_name}{output.CliFormat.ENDC}{output.CliFormat.BLACK} ➔ {_destination_client}{output.CliFormat.ENDC}',

            True
        )

    return _jump_host_channel




1			#!/usr/bin/env python3
2			# -- coding: future_fstrings --
3
4			"""
5			Client script
6			"""
7
8			import sys
9			import paramiko
10			from db_sync_tool.utility import mode, system, helper, output
11
12			ssh_client_origin = None
			0 ignored issues – show Coding Style Naming introduced 2020-11-27 11:55 UTC by Report Bug Copy Issue Report Constant name "ssh_client_origin" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
13			ssh_client_target = None
			0 ignored issues – show Coding Style Naming introduced 2020-11-27 11:55 UTC by Report Bug Copy Issue Report Constant name "ssh_client_target" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
14			additional_ssh_clients = []
15
16			default_timeout = 600
			0 ignored issues – show Coding Style Naming introduced 2021-06-21 15:10 UTC by Report Bug Copy Issue Report Constant name "default_timeout" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
17
18
19			def load_ssh_client_origin():
20			"""
21			Loading the origin ssh client
22			:return:
23			"""
24			global ssh_client_origin
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Usage of the `global` statement should be avoided. Usage of `global` can make code hard to read and test, its usage is generally not recommended unless you are dealing with legacy code. Loading history... Coding Style Naming introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Constant name "ssh_client_origin" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
25			ssh_client_origin = load_ssh_client(mode.Client.ORIGIN)
26			helper.run_script(mode.Client.ORIGIN, 'before')
27
28
29			def load_ssh_client_target():
30			"""
31			Loading the target ssh client
32			:return:
33			"""
34			global ssh_client_target
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Usage of the `global` statement should be avoided. Usage of `global` can make code hard to read and test, its usage is generally not recommended unless you are dealing with legacy code. Loading history... Coding Style Naming introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Constant name "ssh_client_target" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
35			ssh_client_target = load_ssh_client(mode.Client.TARGET)
36			helper.run_script(mode.Client.TARGET, 'before')
37
38
39			def load_ssh_client(ssh):
40			"""
41			Initializing the given ssh client
42			:param ssh: String
43			:return:
44			"""
45			_host_name = helper.get_ssh_host_name(ssh, True)
46			_ssh_client = paramiko.SSHClient()
47			_ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
48
49			_ssh_port = system.config[ssh]['port'] if 'port' in system.config[ssh] else 22
50			_ssh_key = None
51			_ssh_password = None
52
53			# Check authentication
54			if 'ssh_key' in system.config[ssh]:
55			_authentication_method = f'{output.CliFormat.BLACK} - ' \
56			f'(authentication: key){output.CliFormat.ENDC}'
57			_ssh_key = system.config[ssh]['ssh_key']
58			elif 'password' in system.config[ssh]:
59			_authentication_method = f'{output.CliFormat.BLACK} - ' \
60			f'authentication: password){output.CliFormat.ENDC}'
61			_ssh_password = system.config[ssh]['password']
62			elif 'ssh_agent' in system.config:
63			_authentication_method = f'{output.CliFormat.BLACK} - ' \
64			f'(authentication: key){output.CliFormat.ENDC}'
65			else:
66			sys.exit(
67			output.message(
68			output.Subject.ERROR,
69			'Missing SSH authentication. Neither ssh key nor ssh password given.',
70			False
71			)
72			)
73
74			# Try to connect to remote client via paramiko
75			try:
76			_ssh_client.connect(hostname=system.config[ssh]['host'],
77			username=system.config[ssh]['user'],
78			key_filename=_ssh_key,
79			password=_ssh_password,
80			port=_ssh_port,
81			compress=True,
82			timeout=default_timeout,
83			sock=get_jump_host_channel(ssh))
84			#
85			# Workaround for long-lasting requests
86			# https://stackoverflow.com/questions/50009688/python-paramiko-ssh-session-not-active-after-being-idle-for-many-hours
87			#
88			_ssh_client.get_transport().set_keepalive(60)
89
90			except paramiko.ssh_exception.AuthenticationException:
91			sys.exit(
92			output.message(
93			output.Subject.ERROR,
94			f'SSH authentication for {_host_name} failed',
95			False
96			)
97			)
98
99			output.message(
100			output.host_to_subject(ssh),
101			f'Initialize remote SSH connection {_host_name}{_authentication_method}',
102			True
103			)
104
105			return _ssh_client
106
107
108			def close_ssh_clients():
109			"""
110			Closing ssh client sessions
111			:return:
112			"""
113			helper.run_script(mode.Client.ORIGIN, 'after')
114			if not ssh_client_origin is None:
115			ssh_client_origin.close()
116
117			helper.run_script(mode.Client.TARGET, 'after')
118			if not ssh_client_target is None:
119			ssh_client_target.close()
120
121			for additional_ssh_client in additional_ssh_clients:
122			additional_ssh_client.close()
123
124			helper.run_script(script='after')
125
126
127			def get_jump_host_channel(client):
128			"""
129			Provide an optional transport channel for a SSH jump host client
130			https://gist.github.com/tintoy/443c42ea3865680cd624039c4bb46219
131			:param client:
132			:return:
133			"""
134			_jump_host_channel = None
135			if 'jump_host' in system.config[client]:
136			# prepare jump host config
137			_jump_host_client = paramiko.SSHClient()
138			_jump_host_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
139
140			_jump_host_host = system.config[client]['jump_host']['host']
141			_jump_host_user = system.config[client]['jump_host']['user'] if 'user' in system.config[client]['jump_host'] else system.config[client]['user']
			0 ignored issues – show Coding Style introduced 2022-04-10 17:19 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (151/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
142
143			if 'ssh_key' in system.config[client]['jump_host']:
144			_jump_host_ssh_key = system.config[client]['jump_host']['ssh_key']
145			elif 'ssh_key' in system.config[client]:
146			_jump_host_ssh_key = system.config[client]['ssh_key']
147			else:
148			_jump_host_ssh_key = None
149
150			if 'port' in system.config[client]['jump_host']:
151			_jump_host_port = system.config[client]['jump_host']['port']
152			elif 'port' in system.config[client]:
153			_jump_host_port = system.config[client]['port']
154			else:
155			_jump_host_port = 22
156
157			# connect to the jump host
158			_jump_host_client.connect(
159			hostname=_jump_host_host,
160			username=_jump_host_user,
161			key_filename=_jump_host_ssh_key,
162			password=system.config[client]['jump_host']['password'] if 'password' in system.config[client]['jump_host'] else None,
			0 ignored issues – show Coding Style introduced 2022-04-10 17:19 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (130/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
163			port=_jump_host_port,
164			compress=True,
165			timeout=default_timeout
166			)
167
168			global additional_ssh_clients
			0 ignored issues – show Coding Style introduced 2022-04-10 17:19 UTC by Report Bug Copy Issue Report Usage of the `global` statement should be avoided. Usage of `global` can make code hard to read and test, its usage is generally not recommended unless you are dealing with legacy code. Loading history... Coding Style Naming introduced 2022-04-10 17:19 UTC by Report Bug Copy Issue Report Constant name "additional_ssh_clients" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
169			additional_ssh_clients.append(_jump_host_client)
170
171			# open the necessary channel
172			_jump_host_transport = _jump_host_client.get_transport()
173			_jump_host_channel = _jump_host_transport.open_channel(
174			'direct-tcpip',
175			dest_addr=(system.config[client]['host'], 22),
176			src_addr=(system.config[client]['jump_host']['private'] if 'private' in system.config[client]['jump_host'] else system.config[client]['jump_host']['host'], 22)
			0 ignored issues – show Coding Style introduced 2022-04-10 17:19 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (171/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
177			)
178
179			# print information
180			_destination_client = helper.get_ssh_host_name(client, minimal=True)
181			_jump_host_name = system.config[client]['jump_host']['name'] if 'name' in system.config[client]['jump_host'] else _jump_host_host
			0 ignored issues – show Coding Style introduced 2022-04-10 17:19 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (137/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
182			output.message(
183			output.host_to_subject(client),
184			f'Initialize remote SSH jump host {output.CliFormat.BLACK}local ➔ {output.CliFormat.BOLD}{_jump_host_name}{output.CliFormat.ENDC}{output.CliFormat.BLACK} ➔ {_destination_client}{output.CliFormat.ENDC}',
			0 ignored issues – show Coding Style introduced 2022-04-10 17:19 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (214/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
185			True
186			)
187
188			return _jump_host_channel
189
			0 ignored issues – show coding-style introduced 2022-04-10 17:19 UTC by Report Bug Copy Issue Report Trailing newlines Loading history...
190

jackd248 / db-sync-tool

load_ssh_client_target() A last analyzed 2025-04-09 15:26 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

load_ssh_client_target() A
last analyzed 2025-04-09 15:26 UTC