docker_actions.create_network() - Code Metrics - Inspection of "Test with pep8 + better options" - stakkr-org/stakkr - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 0fd795...73442f )

by Emmanuel

created 2018-08-04 12:52 UTC

docker_actions.create_network() A

↳ Parent: docker_actions

Complexity

Conditions

Size

Total Lines	7
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
eloc	4
nop	1
dl	0
loc	7
ccs	4
cts	4
cp	1
crap	2
rs	10
c	0
b	0
f	0

"""Docker functions to get info about containers"""

from docker.errors import NotFound, NullResource
__st__ = {'cts_info': dict(), 'running_cts': 0}


def add_container_to_network(container: str, network: str):
    """Attach a container to a network"""

    if _container_in_network(container, network) is True:
        return False

    docker_network = get_client().networks.get(network)
    docker_network.connect(container)

    return True


def block_ct_ports(service: str, ports: list, project_name: str) -> tuple:
    """Run iptables commands to block a list of port on a specific container"""

    try:
        container = get_client().containers.get(get_ct_item(service, 'id'))
    except (LookupError, NullResource):
        return (False, '{} is not started, no port to block'.format(service))

    status, iptables = container.exec_run(['which', 'iptables'])
    iptables = iptables.decode().strip()
    if iptables == '':
        return (True, "Can't block ports on {}, is iptables installed ?".format(service))

    _allow_contact_subnet(project_name, container)

    # Now for each port, add an iptable rule
    for port in ports:
        rule = ['OUTPUT', '-p', 'tcp', '--dport', port, '-j', 'REJECT']
        try:
            container.exec_run([iptables, '-D'] + rule)
        finally:
            container.exec_run([iptables, '-A'] + rule)

    return (False, 'Blocked ports {} on container {}'.format(', '.join(ports), service))


def check_cts_are_running(project_name: str):
    """Throws an error if cts are not running"""

    get_running_containers(project_name)
    if __st__['running_cts'] is 0:
        raise SystemError('Have you started your server with the start action ?')


def container_running(container: str):
    """Returns True if the container is running else False"""

    try:
        return get_api_client().inspect_container(container)['State']['Running']
    except (NotFound, NullResource):
        return False


def create_network(network: str):
    """Create a Network"""

    if network_exists(network):
        return False

    return get_client().networks.create(network, driver='bridge').id


def get_api_client():
    """Returns the API client or initialize it"""

    if 'api_client' not in __st__:

        from docker import APIClient, utils
        params = utils.kwargs_from_env()
        base_url = None if 'base_url' not in params else params['base_url']
        tls = None if 'tls' not in params else params['tls']

        __st__['api_client'] = APIClient(base_url=base_url, tls=tls)

    return __st__['api_client']


def get_client():
    """Returns the client or initialize it"""

    if 'client' not in __st__:

        from docker import client
        __st__['client'] = client.from_env()

    return __st__['client']


def get_ct_item(compose_name: str, item_name: str):
    """Get a value from a container, such as name or IP"""

    if 'cts_info' not in __st__:
        raise LookupError('Before getting an info from a ct, run check_cts_are_running()')

    for ct_id, ct_data in __st__['cts_info'].items():
        if ct_data['compose_name'] == compose_name:
            return ct_data[item_name]

    return ''


def get_ct_name(container: str):
    """Returns the system name of a container, generated by docker-compose"""

    ct_name = get_ct_item(container, 'name')
    if ct_name == '':
        raise LookupError('{} does not seem to be started ...'.format(container))

    return ct_name


def get_network_name(project_name: str):
    """Find the full network name"""

    try:
        guessed_network_name = project_name.replace('-', '') + '_stakkr'
        network = get_client().networks.get(guessed_network_name)
    except NotFound:
        raise RuntimeError("Couldn't identify network (check your project name)")

    return network.name


def get_subnet(project_name: str):
    """Find the subnet of the current project"""

    network_name = get_network_name(project_name)
    network_info = get_client().networks.get(network_name).attrs

    return network_info['IPAM']['Config'][0]['Subnet'].split('/')[0]


def get_switch_ip():
    """find the main docker daemon IP to add routes"""

    import socket

    cmd = r"""/bin/sh -c "ip addr show hvint0 | grep -oE '([0-9]{1,3}\.){3}[0-9]{1,3}'" """
    res = get_client().containers.run(
        'alpine', remove=True, tty=True, privileged=True,
        network_mode='host', pid_mode='host', command=cmd)
    ip_addr = res.strip().decode()

    try:
        socket.inet_aton(ip_addr)
        return ip_addr
    except socket.error:
        raise ValueError('{} is not a valid ip, check docker is running')


def get_running_containers(project_name: str) -> tuple:
    """Get the number of running containers and theirs details for the current stakkr instance"""

    from requests.exceptions import ConnectionError

    filters = {
        'name': '{}_'.format(project_name),
        'status': 'running',
        'network': '{}_stakkr'.format(project_name).replace('-', '')}

    try:
        cts = get_client().containers.list(filters=filters)
    except ConnectionError:
        raise ConnectionError('Make sure docker is installed and running')

    __st__['cts_info'] = dict()
    for container in cts:
        container_info = _extract_container_info(project_name, container.id)
        __st__['cts_info'][container_info['name']] = container_info

    __st__['running_cts'] = len(cts)

    return (__st__['running_cts'], __st__['cts_info'])



def get_running_containers_name(project_name: str) -> list:
    """Get a list of compose names of running containers for the current stakkr instance"""

    cts = get_running_containers(project_name)[1]

    return sorted([ct_data['compose_name'] for docker_name, ct_data in cts.items()])


def guess_shell(container: str) -> str:
    """By searching for binaries, guess what could be the primary shell available"""

    container = get_client().containers.get(container)

    cmd = 'which -a bash sh'
    status, shells = container.exec_run(cmd, stdout=True, stderr=False)
    shells = shells.splitlines()
    if b'/bin/bash' in shells:
        return '/bin/bash'
    elif b'/bin/sh' in shells:
        return '/bin/sh'

    raise EnvironmentError('Could not find a shell for that container')


def network_exists(network: str):
    """True if a network exists in docker, else False"""

    try:
        get_client().networks.get(network)
        return True
    except NotFound:
        return False


def _allow_contact_subnet(project_name: str, container: str) -> None:
    status, iptables = container.exec_run(['which', 'iptables'])
    iptables = iptables.decode().strip()
    if iptables == '':
        return False

    subnet = get_subnet(project_name) + '/24'
    # Allow internal network
    try:
        container.exec_run([iptables, '-D', 'OUTPUT', '-d', subnet, '-j', 'ACCEPT'])
    finally:
        container.exec_run([iptables, '-A', 'OUTPUT', '-d', subnet, '-j', 'ACCEPT'])


def _extract_container_info(project_name: str, ct_id: str):
    """Get a hash of info about a container : name, ports, image, ip ..."""

    try:
        ct_data = get_api_client().inspect_container(ct_id)
    except NotFound:
        return None

    cts_info = {
        'id': ct_id,
        'name': ct_data['Name'].lstrip('/'),
        'compose_name': ct_data['Config']['Labels']['com.docker.compose.service'],
        'ports': _extract_host_ports(ct_data),
        'image': ct_data['Config']['Image'],
        'traefik_host': _get_traefik_host(ct_data['Config']['Labels']),
        'ip': _get_ip_from_networks(project_name, ct_data['NetworkSettings']['Networks']),
        'running': ct_data['State']['Running']
        }

    return cts_info


def _extract_host_ports(config: list):
    ports = []
    for ct_port, host_ports in config['HostConfig']['PortBindings'].items():
        ports += [host_port['HostPort'] for host_port in host_ports]

    return ports


def _get_ip_from_networks(project_name: str, networks: list):
    """Get a list of IPs for a network"""

    project_name = project_name.replace('-', '')
    network_settings = {}
    if '{}_stakkr'.format(project_name) in networks:
        network_settings = networks['{}_stakkr'.format(project_name)]

    return network_settings['IPAddress'] if 'IPAddress' in network_settings else ''


def _container_in_network(container: str, expected_network: str):
    """Returns True if a container is in a network else false. Used by add_container_to_network"""

    try:
        ct_data = get_api_client().inspect_container(container)
    except NotFound:
        raise LookupError('Container {} does not seem to exist'.format(container))

    for connected_network in ct_data['NetworkSettings']['Networks'].keys():
        if connected_network == expected_network:
            return True

    return False


def _get_traefik_host(labels: list):
    if 'traefik.frontend.rule' not in labels:
        return 'No traefik rule'

    rules = labels['traefik.frontend.rule'].split(':')

    return rules[1]


1		"""Docker functions to get info about containers"""
2
3	1	from docker.errors import NotFound, NullResource
4	1	__st__ = {'cts_info': dict(), 'running_cts': 0}
5
6
7	1	def add_container_to_network(container: str, network: str):
8		"""Attach a container to a network"""
9
10	1	if _container_in_network(container, network) is True:
11	1	return False
12
13	1	docker_network = get_client().networks.get(network)
14	1	docker_network.connect(container)
15
16	1	return True
17
18
19	1	def block_ct_ports(service: str, ports: list, project_name: str) -> tuple:
20		"""Run iptables commands to block a list of port on a specific container"""
21
22	1	try:
23	1	container = get_client().containers.get(get_ct_item(service, 'id'))
24		except (LookupError, NullResource):
25		return (False, '{} is not started, no port to block'.format(service))
26
27	1	status, iptables = container.exec_run(['which', 'iptables'])
28	1	iptables = iptables.decode().strip()
29	1	if iptables == '':
30		return (True, "Can't block ports on {}, is iptables installed ?".format(service))
31
32	1	_allow_contact_subnet(project_name, container)
33
34		# Now for each port, add an iptable rule
35	1	for port in ports:
36	1	rule = ['OUTPUT', '-p', 'tcp', '--dport', port, '-j', 'REJECT']
37	1	try:
38	1	container.exec_run([iptables, '-D'] + rule)
39		finally:
40	1	container.exec_run([iptables, '-A'] + rule)
41
42	1	return (False, 'Blocked ports {} on container {}'.format(', '.join(ports), service))
43
44
45	1	def check_cts_are_running(project_name: str):
46		"""Throws an error if cts are not running"""
47
48	1	get_running_containers(project_name)
49	1	if __st__['running_cts'] is 0:
50	1	raise SystemError('Have you started your server with the start action ?')
51
52
53	1	def container_running(container: str):
54		"""Returns True if the container is running else False"""
55
56	1	try:
57	1	return get_api_client().inspect_container(container)['State']['Running']
58	1	except (NotFound, NullResource):
59	1	return False
60
61
62	1	def create_network(network: str):
63		"""Create a Network"""
64
65	1	if network_exists(network):
66	1	return False
67
68	1	return get_client().networks.create(network, driver='bridge').id
69
70
71	1	def get_api_client():
72		"""Returns the API client or initialize it"""
73
74	1	if 'api_client' not in __st__:
		0 ignored issues – show Comprehensibility Best Practice introduced 2018-07-30 10:17 UTC by Report Bug Copy Issue Report The variable `__st__` does not seem to be defined. Loading history...
75	1	from docker import APIClient, utils
76	1	params = utils.kwargs_from_env()
77	1	base_url = None if 'base_url' not in params else params['base_url']
78	1	tls = None if 'tls' not in params else params['tls']
79
80	1	__st__['api_client'] = APIClient(base_url=base_url, tls=tls)
81
82	1	return __st__['api_client']
83
84
85	1	def get_client():
86		"""Returns the client or initialize it"""
87
88	1	if 'client' not in __st__:
		0 ignored issues – show Comprehensibility Best Practice introduced 2018-07-30 10:17 UTC by Report Bug Copy Issue Report The variable `__st__` does not seem to be defined. Loading history...
89	1	from docker import client
90	1	__st__['client'] = client.from_env()
91
92	1	return __st__['client']
93
94
95	1	def get_ct_item(compose_name: str, item_name: str):
96		"""Get a value from a container, such as name or IP"""
97
98	1	if 'cts_info' not in __st__:
99		raise LookupError('Before getting an info from a ct, run check_cts_are_running()')
100
101	1	for ct_id, ct_data in __st__['cts_info'].items():
102	1	if ct_data['compose_name'] == compose_name:
103	1	return ct_data[item_name]
104
105		return ''
106
107
108	1	def get_ct_name(container: str):
109		"""Returns the system name of a container, generated by docker-compose"""
110
111	1	ct_name = get_ct_item(container, 'name')
112	1	if ct_name == '':
113		raise LookupError('{} does not seem to be started ...'.format(container))
114
115	1	return ct_name
116
117
118	1	def get_network_name(project_name: str):
119		"""Find the full network name"""
120
121	1	try:
122	1	guessed_network_name = project_name.replace('-', '') + '_stakkr'
123	1	network = get_client().networks.get(guessed_network_name)
124		except NotFound:
125		raise RuntimeError("Couldn't identify network (check your project name)")
126
127	1	return network.name
128
129
130	1	def get_subnet(project_name: str):
131		"""Find the subnet of the current project"""
132
133	1	network_name = get_network_name(project_name)
134	1	network_info = get_client().networks.get(network_name).attrs
135
136	1	return network_info['IPAM']['Config'][0]['Subnet'].split('/')[0]
137
138
139	1	def get_switch_ip():
140		"""find the main docker daemon IP to add routes"""
141
142		import socket
143
144		cmd = r"""/bin/sh -c "ip addr show hvint0 \| grep -oE '([0-9]{1,3}\.){3}[0-9]{1,3}'" """
145		res = get_client().containers.run(
146		'alpine', remove=True, tty=True, privileged=True,
147		network_mode='host', pid_mode='host', command=cmd)
148		ip_addr = res.strip().decode()
149
150		try:
151		socket.inet_aton(ip_addr)
152		return ip_addr
153		except socket.error:
154		raise ValueError('{} is not a valid ip, check docker is running')
155
156
157	1	def get_running_containers(project_name: str) -> tuple:
158		"""Get the number of running containers and theirs details for the current stakkr instance"""
159
160	1	from requests.exceptions import ConnectionError
161
162	1	filters = {
163		'name': '{}_'.format(project_name),
164		'status': 'running',
165		'network': '{}_stakkr'.format(project_name).replace('-', '')}
166
167	1	try:
168	1	cts = get_client().containers.list(filters=filters)
169		except ConnectionError:
170		raise ConnectionError('Make sure docker is installed and running')
171
172	1	__st__['cts_info'] = dict()
173	1	for container in cts:
174	1	container_info = _extract_container_info(project_name, container.id)
175	1	__st__['cts_info'][container_info['name']] = container_info
176
177	1	__st__['running_cts'] = len(cts)
178
179	1	return (__st__['running_cts'], __st__['cts_info'])
		0 ignored issues – show Comprehensibility Best Practice introduced 2018-07-30 10:17 UTC by Report Bug Copy Issue Report The variable `__st__` does not seem to be defined. Loading history...
180
181
182	1	def get_running_containers_name(project_name: str) -> list:
183		"""Get a list of compose names of running containers for the current stakkr instance"""
184
185	1	cts = get_running_containers(project_name)[1]
186
187	1	return sorted([ct_data['compose_name'] for docker_name, ct_data in cts.items()])
188
189
190	1	def guess_shell(container: str) -> str:
191		"""By searching for binaries, guess what could be the primary shell available"""
192
193	1	container = get_client().containers.get(container)
194
195	1	cmd = 'which -a bash sh'
196	1	status, shells = container.exec_run(cmd, stdout=True, stderr=False)
197	1	shells = shells.splitlines()
198	1	if b'/bin/bash' in shells:
199	1	return '/bin/bash'
200	1	elif b'/bin/sh' in shells:
201	1	return '/bin/sh'
202
203	1	raise EnvironmentError('Could not find a shell for that container')
204
205
206	1	def network_exists(network: str):
207		"""True if a network exists in docker, else False"""
208
209	1	try:
210	1	get_client().networks.get(network)
211	1	return True
212	1	except NotFound:
213	1	return False
214
215
216	1	def _allow_contact_subnet(project_name: str, container: str) -> None:
217	1	status, iptables = container.exec_run(['which', 'iptables'])
218	1	iptables = iptables.decode().strip()
219	1	if iptables == '':
220		return False
221
222	1	subnet = get_subnet(project_name) + '/24'
223		# Allow internal network
224	1	try:
225	1	container.exec_run([iptables, '-D', 'OUTPUT', '-d', subnet, '-j', 'ACCEPT'])
226		finally:
227	1	container.exec_run([iptables, '-A', 'OUTPUT', '-d', subnet, '-j', 'ACCEPT'])
228
229
230	1	def _extract_container_info(project_name: str, ct_id: str):
231		"""Get a hash of info about a container : name, ports, image, ip ..."""
232
233	1	try:
234	1	ct_data = get_api_client().inspect_container(ct_id)
235	1	except NotFound:
236	1	return None
237
238	1	cts_info = {
239		'id': ct_id,
240		'name': ct_data['Name'].lstrip('/'),
241		'compose_name': ct_data['Config']['Labels']['com.docker.compose.service'],
242		'ports': _extract_host_ports(ct_data),
243		'image': ct_data['Config']['Image'],
244		'traefik_host': _get_traefik_host(ct_data['Config']['Labels']),
245		'ip': _get_ip_from_networks(project_name, ct_data['NetworkSettings']['Networks']),
246		'running': ct_data['State']['Running']
247		}
248
249	1	return cts_info
250
251
252	1	def _extract_host_ports(config: list):
253	1	ports = []
254	1	for ct_port, host_ports in config['HostConfig']['PortBindings'].items():
255		ports += [host_port['HostPort'] for host_port in host_ports]
256
257	1	return ports
258
259
260	1	def _get_ip_from_networks(project_name: str, networks: list):
261		"""Get a list of IPs for a network"""
262
263	1	project_name = project_name.replace('-', '')
264	1	network_settings = {}
265	1	if '{}_stakkr'.format(project_name) in networks:
266	1	network_settings = networks['{}_stakkr'.format(project_name)]
267
268	1	return network_settings['IPAddress'] if 'IPAddress' in network_settings else ''
269
270
271	1	def _container_in_network(container: str, expected_network: str):
272		"""Returns True if a container is in a network else false. Used by add_container_to_network"""
273
274	1	try:
275	1	ct_data = get_api_client().inspect_container(container)
276	1	except NotFound:
277	1	raise LookupError('Container {} does not seem to exist'.format(container))
278
279	1	for connected_network in ct_data['NetworkSettings']['Networks'].keys():
280	1	if connected_network == expected_network:
281	1	return True
282
283	1	return False
284
285
286	1	def _get_traefik_host(labels: list):
287	1	if 'traefik.frontend.rule' not in labels:
288	1	return 'No traefik rule'
289
290	1	rules = labels['traefik.frontend.rule'].split(':')
291
292		return rules[1]
293

stakkr-org / stakkr

Push — master ( 0fd795...73442f )

docker_actions.create_network() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like