Like allow_ports, but it will append to the
    existing entry instead of replacing it.
    Takes a single port instead of a list of ports.

    CLI Example:

    .. code-block:: bash

        salt '*' csf.allow_port 22 proto='tcp' direction='in'
513677
def allow_port(port, proto='tcp', direction='both'):
    '''
    
    '''

    ports = get_ports(proto=proto, direction=direction)
    direction = direction.upper()
    _validate_direction_and_proto(direction, proto)
    directions = build_directions(direction)
    results = []
    for direction in directions:
        _ports = ports[direction]
        _ports.append(port)
        results += allow_ports(_ports, proto=proto, direction=direction)
    return results
696698
def add_bridge_port(name, port, promisc=False):
    '''  '''
    log('Adding port {} to bridge {}'.format(port, name))
    subprocess.check_call(["ovs-vsctl", "--", "--may-exist", "add-port",
                           name, port])
    subprocess.check_call(["ip", "link", "set", port, "up"])
    if promisc:
        subprocess.check_call(["ip", "link", "set", port, "promisc", "on"])
    else:
        subprocess.check_call(["ip", "link", "set", port, "promisc", "off"])
737055
def build_ports_dict(nsg, direction_key, ip_protocol):
        
        rules = nsg['properties']['securityRules']
        rules = sorted(rules, key=lambda k: k['properties']['priority'])
        ports = {}

        for rule in rules:
            # Skip rules with different direction
            if not StringUtils.equal(direction_key, rule['properties']['direction']):
                continue

            # Check the protocol: possible values are 'TCP', 'UDP', '*' (both)
            # Skip only if rule and ip_protocol are 'TCP'/'UDP' pair.
            protocol = rule['properties']['protocol']
            if not StringUtils.equal(protocol, "*") and \
               not StringUtils.equal(ip_protocol, "*") and \
               not StringUtils.equal(protocol, ip_protocol):
                continue

            IsAllowed = StringUtils.equal(rule['properties']['access'], 'allow')
            ports_set = PortsRangeHelper.get_ports_set_from_rule(rule)

            for p in ports_set:
                if p not in ports:
                    ports[p] = IsAllowed

        return ports
785086
def select_random(ports=None, exclude_ports=None):
    
    if ports is None:
        ports = available_good_ports()

    if exclude_ports is None:
        exclude_ports = set()

    ports.difference_update(set(exclude_ports))

    for port in random.sample(ports, min(len(ports), 100)):
        if not port_is_used(port):
            return port
    raise PortForException("Can't select a port")
1060950
def set_ssh_port(port=22):
    '''
    
    '''
    _current = global_settings()

    if _current['Global Settings']['SSH_PORT']['VALUE'] == port:
        return True

    _xml = """<RIBCL VERSION="2.0">
                <LOGIN USER_LOGIN="adminname" PASSWORD="password">
                  <RIB_INFO MODE="write">
                    <MOD_GLOBAL_SETTINGS>
                       <SSH_PORT value="{0}"/>
                    </MOD_GLOBAL_SETTINGS>
                  </RIB_INFO>
                </LOGIN>
              </RIBCL>""".format(port)

    return __execute_cmd('Configure_SSH_Port', _xml)
554506
def add_port(self, port):
        
        self.ports.append(port)
        if port.io_type not in self.port_seqs:
            self.port_seqs[port.io_type] = 0
        self.port_seqs[port.io_type] += 1
        port.sequence = self.port_seqs[port.io_type]
        return self
863278
def _port_op(op_name, port, protocol="TCP"):
    """"""
    _args = [op_name]
    icmp = protocol.upper() == "ICMP"
    if icmp:
        _args.append(protocol)
    else:
        _args.append('{}/{}'.format(port, protocol))
    try:
        subprocess.check_call(_args)
    except subprocess.CalledProcessError:
        # Older Juju pre 2.3 doesn't support ICMP
        # so treat it as a no-op if it fails.
        if not icmp:
            raise
326604
def pick_free_port(hostname=REDIRECT_HOST, port=0):
    
    import socket
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        s.bind((hostname, port))  # port=0 finds an open port
    except OSError as e:
        log.warning("Could not bind to %s:%s %s", hostname, port, e)
        if port == 0:
            print('Unable to find an open port for authentication.')
            raise AuthenticationException(e)
        else:
            return pick_free_port(hostname, 0)
    addr, port = s.getsockname()
    s.close()
    return port
548451
def p_ioport(self, p):
        ''
        p[0] = self.create_ioport(p[1], p[2], lineno=p.lineno(2))
        p.set_lineno(0, p.lineno(1))
306708
public static function allowCors()
    {
        $origin = filter_input(INPUT_SERVER, 'HTTP_ORIGIN');
        $protocol = filter_input(INPUT_SERVER, 'REQUEST_SCHEME') ?? 'http';
        $host = preg_replace("/^http(s)?:\/{2}/", "", $origin);
        $allowedHosts = cfg::ALLOW_CORS;
        $allowedHosts[] = cfg::HOST_DEV;

        if (in_array($host, $allowedHosts)) {
            header("Access-Control-Allow-Origin: " . $protocol . "://" . $host);
        }
    }
631770
def drop_port(self, port, interface=None, subnet=None):
        
        args = {
            'port': port,
            'interface': interface,
            'subnet': subnet,
        }
        self._port_chk.check(args)

        return self._client.json('nft.drop_port', args)
1230832
def input_port(name)
      raise NoSuchPortError.new("input port \"#{name}\" not found") unless input_ports.has_key?(name)
      input_ports[name]
    end
223850
def allocate_tcp_port():
    """"""
    # We want to bind() the socket but not listen(). Twisted (in
    # tcp.Port.createInternetSocket) would do several other things:
    # non-blocking, close-on-exec, and SO_REUSEADDR. We don't need
    # non-blocking because we never listen on it, and we don't need
    # close-on-exec because we close it right away. So just add SO_REUSEADDR.
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    if platformType == "posix" and sys.platform != "cygwin":
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port
874246
def ports_as_list(port_str):
    
    if not port_str:
        LOGGER.info("Invalid port value")
        return [None, None]

    if ports_str_check_failed(port_str):
        LOGGER.info("{0}: Port list malformed.")
        return [None, None]

    tcp_list = list()
    udp_list = list()
    ports = port_str.replace(' ', '')
    b_tcp = ports.find("T")
    b_udp = ports.find("U")

    if ports[b_tcp - 1] == ',':
        ports = ports[:b_tcp - 1] + ports[b_tcp:]
    if ports[b_udp - 1] == ',':
        ports = ports[:b_udp - 1] + ports[b_udp:]
    ports = port_str_arrange(ports)

    tports = ''
    uports = ''
    # TCP ports listed first, then UDP ports
    if b_udp != -1 and b_tcp != -1:
        tports = ports[ports.index('T:') + 2:ports.index('U:')]
        uports = ports[ports.index('U:') + 2:]
    # Only UDP ports
    elif b_tcp == -1 and b_udp != -1:
        uports = ports[ports.index('U:') + 2:]
    # Only TCP ports
    elif b_udp == -1 and b_tcp != -1:
        tports = ports[ports.index('T:') + 2:]
    else:
        tports = ports

    if tports:
        for port in tports.split(','):
            if '-' in port:
                tcp_list.extend(port_range_expand(port))
            else:
                tcp_list.append(int(port))
        tcp_list.sort()
    if uports:
        for port in uports.split(','):
            if '-' in port:
                udp_list.extend(port_range_expand(port))
            else:
                udp_list.append(int(port))
        udp_list.sort()

    return (tcp_list, udp_list)
930709
def build_acl_port(self, port, enabled=True):
        " "
        if port is not None:
            if ':' in port:
                range = port.replace(':', ' ')
                acl = "range %(range)s " % {'range': range}
            else:
                acl = "eq %(port)s " % {'port': port}
                if not enabled:
                    acl += "inactive"
            return acl
302113
def get_port_at(self, tile_id, direction):
        
        for port in self.ports:
            if port.tile_id == tile_id and port.direction == direction:
                return port
        port = Port(tile_id, direction, PortType.none)
        self.ports.append(port)
        return port
302144
def _handle_port_request(self, client_data, writer):
        
        try:
            pid = int(client_data)
        except ValueError as error:
            self._client_request_errors += 1
            log.warning('Could not parse request: %s', error)
            return

        log.info('Request on behalf of pid %d.', pid)
        log.info('cmdline: %s', _get_process_command_line(pid))

        if not _should_allocate_port(pid):
            self._denied_allocations += 1
            return

        port = self._port_pool.get_port_for_process(pid)
        if port > 0:
            self._total_allocations += 1
            writer.write('{:d}\n'.format(port).encode('utf-8'))
            log.debug('Allocated port %d to pid %d', port, pid)
        else:
            self._denied_allocations += 1