docker_actions.get_subnet() - Code Metrics - Inspection of "Latest version of 4.x-dev with everything working..." - stakkr-org/stakkr - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Retried Inspection

Branch — v4.0-dev (e44d7e)

by Emmanuel

created 2018-12-08 22:24 UTC

docker_actions.get_subnet() A

↳ Parent: docker_actions

Complexity

Conditions

Size

Total Lines	6
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# coding: utf-8
"""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', str(port), '-j', 'REJECT']
        try:
            container.exec_run([iptables, '-D'] + rule)
        finally:
            container.exec_run([iptables, '-A'] + rule)

    ports_list = ', '.join(map(str, ports))

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


def check_cts_are_running(project_name: str):
    """Throw an error if cts are not running."""
    get_running_containers(project_name)
    if __st__['running_cts'] is 0:
        raise SystemError('Have you started stakkr with the start action ?')


def container_running(container: str):
    """Return 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():
    """Return 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():
    """Return 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):
    """Return 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 = '{}_stakkr'.format(project_name)
        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)}

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

stakkr-org / stakkr

Branch — v4.0-dev (e44d7e)

docker_actions.get_subnet() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like