# This code is part of the Ansible collection community.docker, but is an independent component. # This particular file, and this file only, is based on the Docker SDK for Python (https://github.com/docker/docker-py/) # # Copyright (c) 2016-2022 Docker, Inc. # # It is licensed under the Apache 2.0 license (see LICENSES/Apache-2.0.txt in this collection) # SPDX-License-Identifier: Apache-2.0 # Note that this module util is **PRIVATE** to the collection. It can have breaking changes at any time. # Do not use this from other collections or standalone plugins/modules! from __future__ import annotations import re PORT_SPEC = re.compile( "^" # Match full string "(" # External part r"(\[?(?P[a-fA-F\d.:]+)\]?:)?" # Address r"(?P[\d]*)(-(?P[\d]+))?:" # External range ")?" r"(?P[\d]+)(-(?P[\d]+))?" # Internal range "(?P/(udp|tcp|sctp))?" # Protocol "$" # Match full string ) def add_port_mapping(port_bindings, internal_port, external): if internal_port in port_bindings: port_bindings[internal_port].append(external) else: port_bindings[internal_port] = [external] def add_port(port_bindings, internal_port_range, external_range): if external_range is None: for internal_port in internal_port_range: add_port_mapping(port_bindings, internal_port, None) else: ports = zip(internal_port_range, external_range) for internal_port, external_port in ports: add_port_mapping(port_bindings, internal_port, external_port) def build_port_bindings(ports): port_bindings = {} for port in ports: internal_port_range, external_range = split_port(port) add_port(port_bindings, internal_port_range, external_range) return port_bindings def _raise_invalid_port(port): raise ValueError( f'Invalid port "{port}", should be ' "[[remote_ip:]remote_port[-remote_port]:]" "port[/protocol]" ) def port_range(start, end, proto, randomly_available_port=False): if not start: return start if not end: return [f"{start}{proto}"] if randomly_available_port: return [f"{start}-{end}{proto}"] return [f"{port}{proto}" for port in range(int(start), int(end) + 1)] def split_port(port): if hasattr(port, "legacy_repr"): # This is the worst hack, but it prevents a bug in Compose 1.14.0 # https://github.com/docker/docker-py/issues/1668 # TODO: remove once fixed in Compose stable port = port.legacy_repr() port = str(port) match = PORT_SPEC.match(port) if match is None: _raise_invalid_port(port) parts = match.groupdict() host = parts["host"] proto = parts["proto"] or "" internal = port_range(parts["int"], parts["int_end"], proto) external = port_range(parts["ext"], parts["ext_end"], "", len(internal) == 1) if host is None: if external is not None and len(internal) != len(external): raise ValueError("Port ranges don't match in length") return internal, external else: if not external: external = [None] * len(internal) elif len(internal) != len(external): raise ValueError("Port ranges don't match in length") return internal, [(host, ext_port) for ext_port in external]