stakkr.docker_actions.get_running_containers_names() - Code Metrics - Inspection of "Able to be globally installed" - stakkr-org/stakkr - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — v4.0-dev (e005f1)

by Emmanuel

created 2018-12-13 22:28 UTC

get_running_containers_names() A

↳ Parent: stakkr.docker_actions

Complexity

Conditions

Size

Total Lines	5
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	1
dl	0
loc	5
ccs	3
cts	3
cp	1
crap	1
rs	10
c	0
b	0
f	0

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

stakkr-org / stakkr

Branch — v4.0-dev (e005f1)

get_running_containers_names() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like