block_ct_ports() - Code Metrics - Inspection of "Added new services for stakkr" - stakkr-org/stakkr - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#8)

unknown

created 2018-06-24 18:27 UTC

block_ct_ports() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	20

Importance

Changes	3
Bugs	0	Features	0

Metric	Value
cc	4
c	3
b	0
f	0
dl	0
loc	24
ccs	0
cts	16
cp	0
crap	20
rs	9.304

"""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_subnet(project_name: str):
    """Find the subnet of the current project"""

    network_info = get_client().networks.get(project_name.replace('-', '') + '_stakkr').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')

    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'],
        '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


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

stakkr-org / stakkr

Pull Request — master (#8)

block_ct_ports() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like