Python mở cổng và lắng nghe

54 ví dụ mã Python được tìm thấy liên quan đến " cổng mở". Bạn có thể bỏ phiếu cho những cái bạn thích hoặc bỏ phiếu cho những cái bạn không thích và chuyển đến dự án gốc hoặc tệp nguồn bằng cách nhấp vào các liên kết phía trên mỗi ví dụ

def check_and_open_port_by_upnp[external_port, internal_port, protocol]:
    """open the router's port to enable external connection"""
    request_url = cast_rooter_request[]
    if request_url is None:
        log.debug["node is not in local network protected by a router"]
        return
    soap_url = get_soap_url[request_url]
    internal_client = get_localhost_ip[]
    # check existence
    for mapping in soap_get_mapping[soap_url]:
        if mapping.enabled == 1 and \
                mapping.external_port == external_port and \
                mapping.internal_port == internal_port and \
                mapping.protocol == protocol and \
                mapping.external_client == internal_client:
            return
    # open port
    soap_add_mapping[soap_url, external_port, internal_port, internal_client, protocol]
    log.info[f"open port by upnp {internal_port} -> {external_port}"] 

def setOpenPort[self, portId, oldPortId, quiet=0]:
        openPortUID = EnemyGlobals.OPEN_PORT_DICT.get[portId, '']
        openPortName = PLocalizer.OpenPortNames.get[openPortUID]
        oldPortUID = EnemyGlobals.OPEN_PORT_DICT.get[oldPortId, '']
        oldPortName = PLocalizer.OpenPortNames.get[oldPortUID]
        if openPortName:
            self.openPortLabel.show[]
            self.openPortLabel['text'] = PLocalizer.OpenPortMessage % openPortName
            if not quiet:
                if oldPortId:
                    oldPortMessage = HighSeasGlobals.getPortClosedMessage[oldPortUID]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = oldPortMessage[0] + openPortMessage[0]
                    messageSoundList = [oldPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
                else:
                    allPortMessage = HighSeasGlobals.getInitPortMessage[]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = allPortMessage[0] + openPortMessage[0]
                    messageSoundList = [allPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
            self.disableAnchorButton[]
        else:
            self.openPortLabel.hide[]
            self.hideWrongPort[] 

def open_port[conn]:
    print["Open a port:"]

    example_sec_group = conn.network.create_security_group[
        name='openstacksdk-example-security-group']

    print[example_sec_group]

    example_rule = conn.network.create_security_group_rule[
        security_group_id=example_sec_group.id,
        direction='ingress',
        remote_ip_prefix='0.0.0.0/0',
        protocol='HTTPS',
        port_range_max='443',
        port_range_min='443',
        ethertype='IPv4']

    print[example_rule] 

def wait_for_port_open[self, postmaster, timeout]:
        """Waits until PostgreSQL opens ports."""
        for _ in polling_loop[timeout]:
            if self.cancellable.is_cancelled:
                return False

            if not postmaster.is_running[]:
                logger.error['postmaster is not running']
                self.set_state['start failed']
                return False

            isready = self.pg_isready[]
            if isready != STATE_NO_RESPONSE:
                if isready not in [STATE_REJECT, STATE_RUNNING]:
                    logger.warning["Can't determine PostgreSQL startup status, assuming running"]
                return True

        logger.warning["Timed out waiting for PostgreSQL to start"]
        return False 

def open_port[host, port]:
    """Return whether or not host's port is open.

    Parameters
    ----------
    host : str
        Host IP address or hostname
    port : int
        Port number

    Returns
    -------
    awaitable bool
    """
    try:
        _reader, writer = await asyncio.wait_for[
            asyncio.open_connection[host, port], timeout=2
        ]
        writer.close[]
        await writer.wait_closed[]
        return True
    except Exception:
        pass

    return False 

def check_port_open[port, loop, delay=1]:
    # the "s = socket.socket; s.bind" approach sometimes says a port is in use when it's not
    # this approach replicates aiohttp so should always give the same answer
    for i in range[5, 0, -1]:
        try:
            server = await loop.create_server[asyncio.Protocol[], host=HOST, port=port]
        except OSError as e:
            if e.errno != 98:  # pragma: no cover
                raise
            dft_logger.warning['port %d is already in use, waiting %d...', port, i]
            await asyncio.sleep[delay, loop=loop]
        else:
            server.close[]
            await server.wait_closed[]
            return
    raise AiohttpDevException['The port {} is already is use'.format[port]] 

def get_open_port[ignore_common=False]:
    """
    Get a random TCP port not currently in use by a tracked service.

    :param bool ignore_common: Don't exclude commonly used ports?
    :returns: Port number
    :rtype: int
    """
    data = get[]
    ports = []
    for x in data:
        for y in x.ports:
            ports.append[int[y[1]]]
    if not ignore_common:
        ports = ports + COMMON_PORTS
    r = random.randint[8001, 65534]
    return r if r not in ports else get_open_port[] 

def get_open_port_count[self, flt, limit=None, skip=None]:
        res = self.get[flt, limit=limit, skip=skip]
        return [[{'addr': host['addr'],
                  'starttime': host.get['starttime'],
                  'openports': {'count': host['openports']['count']}}
                 for host in res
                 if host.get['openports', {}].get['count'] is not None],
                len[res]] 

________số 8_______

def try_open_port[self]:
        #self.data.midi_status = 'disconnected'
        self.midi_setting = self.data.settings['user_input']['MIDI_INPUT']['value']
        self.port_index = self.data.midi_port_index
        #print['try open port : midi setting is {}'.format[midi_setting]]
        if self.midi_setting == 'usb':
            self.actions.stop_serial_port_process[]
            self.open_this_port_and_start_listening['20']
        elif self.midi_setting == 'serial':
            self.actions.create_serial_port_process[]
            self.open_this_port_and_start_listening['serial']
        else:
            self.actions.stop_serial_port_process[]     
            self.data.midi_status = 'disconnected'
        self.root.after[1000, self.try_open_port] 

def setOpenPort[self, portId, oldPortId, quiet=0]:
        openPortUID = EnemyGlobals.OPEN_PORT_DICT.get[portId, '']
        openPortName = PLocalizer.OpenPortNames.get[openPortUID]
        oldPortUID = EnemyGlobals.OPEN_PORT_DICT.get[oldPortId, '']
        oldPortName = PLocalizer.OpenPortNames.get[oldPortUID]
        if openPortName:
            self.openPortLabel.show[]
            self.openPortLabel['text'] = PLocalizer.OpenPortMessage % openPortName
            if not quiet:
                if oldPortId:
                    oldPortMessage = HighSeasGlobals.getPortClosedMessage[oldPortUID]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = oldPortMessage[0] + openPortMessage[0]
                    messageSoundList = [oldPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
                else:
                    allPortMessage = HighSeasGlobals.getInitPortMessage[]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = allPortMessage[0] + openPortMessage[0]
                    messageSoundList = [allPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
            self.disableAnchorButton[]
        else:
            self.openPortLabel.hide[]
            self.hideWrongPort[] 
0

def setOpenPort[self, portId, oldPortId, quiet=0]:
        openPortUID = EnemyGlobals.OPEN_PORT_DICT.get[portId, '']
        openPortName = PLocalizer.OpenPortNames.get[openPortUID]
        oldPortUID = EnemyGlobals.OPEN_PORT_DICT.get[oldPortId, '']
        oldPortName = PLocalizer.OpenPortNames.get[oldPortUID]
        if openPortName:
            self.openPortLabel.show[]
            self.openPortLabel['text'] = PLocalizer.OpenPortMessage % openPortName
            if not quiet:
                if oldPortId:
                    oldPortMessage = HighSeasGlobals.getPortClosedMessage[oldPortUID]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = oldPortMessage[0] + openPortMessage[0]
                    messageSoundList = [oldPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
                else:
                    allPortMessage = HighSeasGlobals.getInitPortMessage[]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = allPortMessage[0] + openPortMessage[0]
                    messageSoundList = [allPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
            self.disableAnchorButton[]
        else:
            self.openPortLabel.hide[]
            self.hideWrongPort[] 
1

def setOpenPort[self, portId, oldPortId, quiet=0]:
        openPortUID = EnemyGlobals.OPEN_PORT_DICT.get[portId, '']
        openPortName = PLocalizer.OpenPortNames.get[openPortUID]
        oldPortUID = EnemyGlobals.OPEN_PORT_DICT.get[oldPortId, '']
        oldPortName = PLocalizer.OpenPortNames.get[oldPortUID]
        if openPortName:
            self.openPortLabel.show[]
            self.openPortLabel['text'] = PLocalizer.OpenPortMessage % openPortName
            if not quiet:
                if oldPortId:
                    oldPortMessage = HighSeasGlobals.getPortClosedMessage[oldPortUID]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = oldPortMessage[0] + openPortMessage[0]
                    messageSoundList = [oldPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
                else:
                    allPortMessage = HighSeasGlobals.getInitPortMessage[]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = allPortMessage[0] + openPortMessage[0]
                    messageSoundList = [allPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
            self.disableAnchorButton[]
        else:
            self.openPortLabel.hide[]
            self.hideWrongPort[] 
2

def setOpenPort[self, portId, oldPortId, quiet=0]:
        openPortUID = EnemyGlobals.OPEN_PORT_DICT.get[portId, '']
        openPortName = PLocalizer.OpenPortNames.get[openPortUID]
        oldPortUID = EnemyGlobals.OPEN_PORT_DICT.get[oldPortId, '']
        oldPortName = PLocalizer.OpenPortNames.get[oldPortUID]
        if openPortName:
            self.openPortLabel.show[]
            self.openPortLabel['text'] = PLocalizer.OpenPortMessage % openPortName
            if not quiet:
                if oldPortId:
                    oldPortMessage = HighSeasGlobals.getPortClosedMessage[oldPortUID]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = oldPortMessage[0] + openPortMessage[0]
                    messageSoundList = [oldPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
                else:
                    allPortMessage = HighSeasGlobals.getInitPortMessage[]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = allPortMessage[0] + openPortMessage[0]
                    messageSoundList = [allPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
            self.disableAnchorButton[]
        else:
            self.openPortLabel.hide[]
            self.hideWrongPort[] 
3

def setOpenPort[self, portId, oldPortId, quiet=0]:
        openPortUID = EnemyGlobals.OPEN_PORT_DICT.get[portId, '']
        openPortName = PLocalizer.OpenPortNames.get[openPortUID]
        oldPortUID = EnemyGlobals.OPEN_PORT_DICT.get[oldPortId, '']
        oldPortName = PLocalizer.OpenPortNames.get[oldPortUID]
        if openPortName:
            self.openPortLabel.show[]
            self.openPortLabel['text'] = PLocalizer.OpenPortMessage % openPortName
            if not quiet:
                if oldPortId:
                    oldPortMessage = HighSeasGlobals.getPortClosedMessage[oldPortUID]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = oldPortMessage[0] + openPortMessage[0]
                    messageSoundList = [oldPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
                else:
                    allPortMessage = HighSeasGlobals.getInitPortMessage[]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = allPortMessage[0] + openPortMessage[0]
                    messageSoundList = [allPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
            self.disableAnchorButton[]
        else:
            self.openPortLabel.hide[]
            self.hideWrongPort[] 
4

def setOpenPort[self, portId, oldPortId, quiet=0]:
        openPortUID = EnemyGlobals.OPEN_PORT_DICT.get[portId, '']
        openPortName = PLocalizer.OpenPortNames.get[openPortUID]
        oldPortUID = EnemyGlobals.OPEN_PORT_DICT.get[oldPortId, '']
        oldPortName = PLocalizer.OpenPortNames.get[oldPortUID]
        if openPortName:
            self.openPortLabel.show[]
            self.openPortLabel['text'] = PLocalizer.OpenPortMessage % openPortName
            if not quiet:
                if oldPortId:
                    oldPortMessage = HighSeasGlobals.getPortClosedMessage[oldPortUID]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = oldPortMessage[0] + openPortMessage[0]
                    messageSoundList = [oldPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
                else:
                    allPortMessage = HighSeasGlobals.getInitPortMessage[]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = allPortMessage[0] + openPortMessage[0]
                    messageSoundList = [allPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
            self.disableAnchorButton[]
        else:
            self.openPortLabel.hide[]
            self.hideWrongPort[] 
5

def setOpenPort[self, portId, oldPortId, quiet=0]:
        openPortUID = EnemyGlobals.OPEN_PORT_DICT.get[portId, '']
        openPortName = PLocalizer.OpenPortNames.get[openPortUID]
        oldPortUID = EnemyGlobals.OPEN_PORT_DICT.get[oldPortId, '']
        oldPortName = PLocalizer.OpenPortNames.get[oldPortUID]
        if openPortName:
            self.openPortLabel.show[]
            self.openPortLabel['text'] = PLocalizer.OpenPortMessage % openPortName
            if not quiet:
                if oldPortId:
                    oldPortMessage = HighSeasGlobals.getPortClosedMessage[oldPortUID]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = oldPortMessage[0] + openPortMessage[0]
                    messageSoundList = [oldPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
                else:
                    allPortMessage = HighSeasGlobals.getInitPortMessage[]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = allPortMessage[0] + openPortMessage[0]
                    messageSoundList = [allPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
            self.disableAnchorButton[]
        else:
            self.openPortLabel.hide[]
            self.hideWrongPort[] 
6

def setOpenPort[self, portId, oldPortId, quiet=0]:
        openPortUID = EnemyGlobals.OPEN_PORT_DICT.get[portId, '']
        openPortName = PLocalizer.OpenPortNames.get[openPortUID]
        oldPortUID = EnemyGlobals.OPEN_PORT_DICT.get[oldPortId, '']
        oldPortName = PLocalizer.OpenPortNames.get[oldPortUID]
        if openPortName:
            self.openPortLabel.show[]
            self.openPortLabel['text'] = PLocalizer.OpenPortMessage % openPortName
            if not quiet:
                if oldPortId:
                    oldPortMessage = HighSeasGlobals.getPortClosedMessage[oldPortUID]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = oldPortMessage[0] + openPortMessage[0]
                    messageSoundList = [oldPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
                else:
                    allPortMessage = HighSeasGlobals.getInitPortMessage[]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = allPortMessage[0] + openPortMessage[0]
                    messageSoundList = [allPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
            self.disableAnchorButton[]
        else:
            self.openPortLabel.hide[]
            self.hideWrongPort[] 
7

def setOpenPort[self, portId, oldPortId, quiet=0]:
        openPortUID = EnemyGlobals.OPEN_PORT_DICT.get[portId, '']
        openPortName = PLocalizer.OpenPortNames.get[openPortUID]
        oldPortUID = EnemyGlobals.OPEN_PORT_DICT.get[oldPortId, '']
        oldPortName = PLocalizer.OpenPortNames.get[oldPortUID]
        if openPortName:
            self.openPortLabel.show[]
            self.openPortLabel['text'] = PLocalizer.OpenPortMessage % openPortName
            if not quiet:
                if oldPortId:
                    oldPortMessage = HighSeasGlobals.getPortClosedMessage[oldPortUID]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = oldPortMessage[0] + openPortMessage[0]
                    messageSoundList = [oldPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
                else:
                    allPortMessage = HighSeasGlobals.getInitPortMessage[]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = allPortMessage[0] + openPortMessage[0]
                    messageSoundList = [allPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
            self.disableAnchorButton[]
        else:
            self.openPortLabel.hide[]
            self.hideWrongPort[] 
8

def setOpenPort[self, portId, oldPortId, quiet=0]:
        openPortUID = EnemyGlobals.OPEN_PORT_DICT.get[portId, '']
        openPortName = PLocalizer.OpenPortNames.get[openPortUID]
        oldPortUID = EnemyGlobals.OPEN_PORT_DICT.get[oldPortId, '']
        oldPortName = PLocalizer.OpenPortNames.get[oldPortUID]
        if openPortName:
            self.openPortLabel.show[]
            self.openPortLabel['text'] = PLocalizer.OpenPortMessage % openPortName
            if not quiet:
                if oldPortId:
                    oldPortMessage = HighSeasGlobals.getPortClosedMessage[oldPortUID]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = oldPortMessage[0] + openPortMessage[0]
                    messageSoundList = [oldPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
                else:
                    allPortMessage = HighSeasGlobals.getInitPortMessage[]
                    openPortMessage = HighSeasGlobals.getPortOpenMessage[openPortUID]
                    messageText = allPortMessage[0] + openPortMessage[0]
                    messageSoundList = [allPortMessage[1], openPortMessage[1]]
                    base.localAvatar.guiMgr.queueInstructionMessage[messageText, messageSoundList, messageCategory=MessageGlobals.MSG_CAT_TELL_PORT]
            self.disableAnchorButton[]
        else:
            self.openPortLabel.hide[]
            self.hideWrongPort[] 
9

def open_port[conn]:
    print["Open a port:"]

    example_sec_group = conn.network.create_security_group[
        name='openstacksdk-example-security-group']

    print[example_sec_group]

    example_rule = conn.network.create_security_group_rule[
        security_group_id=example_sec_group.id,
        direction='ingress',
        remote_ip_prefix='0.0.0.0/0',
        protocol='HTTPS',
        port_range_max='443',
        port_range_min='443',
        ethertype='IPv4']

    print[example_rule] 
0

def open_port[conn]:
    print["Open a port:"]

    example_sec_group = conn.network.create_security_group[
        name='openstacksdk-example-security-group']

    print[example_sec_group]

    example_rule = conn.network.create_security_group_rule[
        security_group_id=example_sec_group.id,
        direction='ingress',
        remote_ip_prefix='0.0.0.0/0',
        protocol='HTTPS',
        port_range_max='443',
        port_range_min='443',
        ethertype='IPv4']

    print[example_rule] 
1

def open_port[conn]:
    print["Open a port:"]

    example_sec_group = conn.network.create_security_group[
        name='openstacksdk-example-security-group']

    print[example_sec_group]

    example_rule = conn.network.create_security_group_rule[
        security_group_id=example_sec_group.id,
        direction='ingress',
        remote_ip_prefix='0.0.0.0/0',
        protocol='HTTPS',
        port_range_max='443',
        port_range_min='443',
        ethertype='IPv4']

    print[example_rule] 
2

def open_port[conn]:
    print["Open a port:"]

    example_sec_group = conn.network.create_security_group[
        name='openstacksdk-example-security-group']

    print[example_sec_group]

    example_rule = conn.network.create_security_group_rule[
        security_group_id=example_sec_group.id,
        direction='ingress',
        remote_ip_prefix='0.0.0.0/0',
        protocol='HTTPS',
        port_range_max='443',
        port_range_min='443',
        ethertype='IPv4']

    print[example_rule] 
3

def open_port[conn]:
    print["Open a port:"]

    example_sec_group = conn.network.create_security_group[
        name='openstacksdk-example-security-group']

    print[example_sec_group]

    example_rule = conn.network.create_security_group_rule[
        security_group_id=example_sec_group.id,
        direction='ingress',
        remote_ip_prefix='0.0.0.0/0',
        protocol='HTTPS',
        port_range_max='443',
        port_range_min='443',
        ethertype='IPv4']

    print[example_rule] 
4

def open_port[conn]:
    print["Open a port:"]

    example_sec_group = conn.network.create_security_group[
        name='openstacksdk-example-security-group']

    print[example_sec_group]

    example_rule = conn.network.create_security_group_rule[
        security_group_id=example_sec_group.id,
        direction='ingress',
        remote_ip_prefix='0.0.0.0/0',
        protocol='HTTPS',
        port_range_max='443',
        port_range_min='443',
        ethertype='IPv4']

    print[example_rule] 
5

def open_port[conn]:
    print["Open a port:"]

    example_sec_group = conn.network.create_security_group[
        name='openstacksdk-example-security-group']

    print[example_sec_group]

    example_rule = conn.network.create_security_group_rule[
        security_group_id=example_sec_group.id,
        direction='ingress',
        remote_ip_prefix='0.0.0.0/0',
        protocol='HTTPS',
        port_range_max='443',
        port_range_min='443',
        ethertype='IPv4']

    print[example_rule] 
6

def open_port[conn]:
    print["Open a port:"]

    example_sec_group = conn.network.create_security_group[
        name='openstacksdk-example-security-group']

    print[example_sec_group]

    example_rule = conn.network.create_security_group_rule[
        security_group_id=example_sec_group.id,
        direction='ingress',
        remote_ip_prefix='0.0.0.0/0',
        protocol='HTTPS',
        port_range_max='443',
        port_range_min='443',
        ethertype='IPv4']

    print[example_rule] 
7

def open_port[conn]:
    print["Open a port:"]

    example_sec_group = conn.network.create_security_group[
        name='openstacksdk-example-security-group']

    print[example_sec_group]

    example_rule = conn.network.create_security_group_rule[
        security_group_id=example_sec_group.id,
        direction='ingress',
        remote_ip_prefix='0.0.0.0/0',
        protocol='HTTPS',
        port_range_max='443',
        port_range_min='443',
        ethertype='IPv4']

    print[example_rule] 
8

def open_port[conn]:
    print["Open a port:"]

    example_sec_group = conn.network.create_security_group[
        name='openstacksdk-example-security-group']

    print[example_sec_group]

    example_rule = conn.network.create_security_group_rule[
        security_group_id=example_sec_group.id,
        direction='ingress',
        remote_ip_prefix='0.0.0.0/0',
        protocol='HTTPS',
        port_range_max='443',
        port_range_min='443',
        ethertype='IPv4']

    print[example_rule] 
9

def wait_for_port_open[self, postmaster, timeout]:
        """Waits until PostgreSQL opens ports."""
        for _ in polling_loop[timeout]:
            if self.cancellable.is_cancelled:
                return False

            if not postmaster.is_running[]:
                logger.error['postmaster is not running']
                self.set_state['start failed']
                return False

            isready = self.pg_isready[]
            if isready != STATE_NO_RESPONSE:
                if isready not in [STATE_REJECT, STATE_RUNNING]:
                    logger.warning["Can't determine PostgreSQL startup status, assuming running"]
                return True

        logger.warning["Timed out waiting for PostgreSQL to start"]
        return False 
0

def wait_for_port_open[self, postmaster, timeout]:
        """Waits until PostgreSQL opens ports."""
        for _ in polling_loop[timeout]:
            if self.cancellable.is_cancelled:
                return False

            if not postmaster.is_running[]:
                logger.error['postmaster is not running']
                self.set_state['start failed']
                return False

            isready = self.pg_isready[]
            if isready != STATE_NO_RESPONSE:
                if isready not in [STATE_REJECT, STATE_RUNNING]:
                    logger.warning["Can't determine PostgreSQL startup status, assuming running"]
                return True

        logger.warning["Timed out waiting for PostgreSQL to start"]
        return False 
1

def wait_for_port_open[self, postmaster, timeout]:
        """Waits until PostgreSQL opens ports."""
        for _ in polling_loop[timeout]:
            if self.cancellable.is_cancelled:
                return False

            if not postmaster.is_running[]:
                logger.error['postmaster is not running']
                self.set_state['start failed']
                return False

            isready = self.pg_isready[]
            if isready != STATE_NO_RESPONSE:
                if isready not in [STATE_REJECT, STATE_RUNNING]:
                    logger.warning["Can't determine PostgreSQL startup status, assuming running"]
                return True

        logger.warning["Timed out waiting for PostgreSQL to start"]
        return False 
2

def wait_for_port_open[self, postmaster, timeout]:
        """Waits until PostgreSQL opens ports."""
        for _ in polling_loop[timeout]:
            if self.cancellable.is_cancelled:
                return False

            if not postmaster.is_running[]:
                logger.error['postmaster is not running']
                self.set_state['start failed']
                return False

            isready = self.pg_isready[]
            if isready != STATE_NO_RESPONSE:
                if isready not in [STATE_REJECT, STATE_RUNNING]:
                    logger.warning["Can't determine PostgreSQL startup status, assuming running"]
                return True

        logger.warning["Timed out waiting for PostgreSQL to start"]
        return False 
3

def wait_for_port_open[self, postmaster, timeout]:
        """Waits until PostgreSQL opens ports."""
        for _ in polling_loop[timeout]:
            if self.cancellable.is_cancelled:
                return False

            if not postmaster.is_running[]:
                logger.error['postmaster is not running']
                self.set_state['start failed']
                return False

            isready = self.pg_isready[]
            if isready != STATE_NO_RESPONSE:
                if isready not in [STATE_REJECT, STATE_RUNNING]:
                    logger.warning["Can't determine PostgreSQL startup status, assuming running"]
                return True

        logger.warning["Timed out waiting for PostgreSQL to start"]
        return False 
4

def wait_for_port_open[self, postmaster, timeout]:
        """Waits until PostgreSQL opens ports."""
        for _ in polling_loop[timeout]:
            if self.cancellable.is_cancelled:
                return False

            if not postmaster.is_running[]:
                logger.error['postmaster is not running']
                self.set_state['start failed']
                return False

            isready = self.pg_isready[]
            if isready != STATE_NO_RESPONSE:
                if isready not in [STATE_REJECT, STATE_RUNNING]:
                    logger.warning["Can't determine PostgreSQL startup status, assuming running"]
                return True

        logger.warning["Timed out waiting for PostgreSQL to start"]
        return False 
5

def wait_for_port_open[self, postmaster, timeout]:
        """Waits until PostgreSQL opens ports."""
        for _ in polling_loop[timeout]:
            if self.cancellable.is_cancelled:
                return False

            if not postmaster.is_running[]:
                logger.error['postmaster is not running']
                self.set_state['start failed']
                return False

            isready = self.pg_isready[]
            if isready != STATE_NO_RESPONSE:
                if isready not in [STATE_REJECT, STATE_RUNNING]:
                    logger.warning["Can't determine PostgreSQL startup status, assuming running"]
                return True

        logger.warning["Timed out waiting for PostgreSQL to start"]
        return False 
6

def wait_for_port_open[self, postmaster, timeout]:
        """Waits until PostgreSQL opens ports."""
        for _ in polling_loop[timeout]:
            if self.cancellable.is_cancelled:
                return False

            if not postmaster.is_running[]:
                logger.error['postmaster is not running']
                self.set_state['start failed']
                return False

            isready = self.pg_isready[]
            if isready != STATE_NO_RESPONSE:
                if isready not in [STATE_REJECT, STATE_RUNNING]:
                    logger.warning["Can't determine PostgreSQL startup status, assuming running"]
                return True

        logger.warning["Timed out waiting for PostgreSQL to start"]
        return False 
7

def wait_for_port_open[self, postmaster, timeout]:
        """Waits until PostgreSQL opens ports."""
        for _ in polling_loop[timeout]:
            if self.cancellable.is_cancelled:
                return False

            if not postmaster.is_running[]:
                logger.error['postmaster is not running']
                self.set_state['start failed']
                return False

            isready = self.pg_isready[]
            if isready != STATE_NO_RESPONSE:
                if isready not in [STATE_REJECT, STATE_RUNNING]:
                    logger.warning["Can't determine PostgreSQL startup status, assuming running"]
                return True

        logger.warning["Timed out waiting for PostgreSQL to start"]
        return False 
8

def wait_for_port_open[self, postmaster, timeout]:
        """Waits until PostgreSQL opens ports."""
        for _ in polling_loop[timeout]:
            if self.cancellable.is_cancelled:
                return False

            if not postmaster.is_running[]:
                logger.error['postmaster is not running']
                self.set_state['start failed']
                return False

            isready = self.pg_isready[]
            if isready != STATE_NO_RESPONSE:
                if isready not in [STATE_REJECT, STATE_RUNNING]:
                    logger.warning["Can't determine PostgreSQL startup status, assuming running"]
                return True

        logger.warning["Timed out waiting for PostgreSQL to start"]
        return False 
9

def open_port[host, port]:
    """Return whether or not host's port is open.

    Parameters
    ----------
    host : str
        Host IP address or hostname
    port : int
        Port number

    Returns
    -------
    awaitable bool
    """
    try:
        _reader, writer = await asyncio.wait_for[
            asyncio.open_connection[host, port], timeout=2
        ]
        writer.close[]
        await writer.wait_closed[]
        return True
    except Exception:
        pass

    return False 
0

def open_port[host, port]:
    """Return whether or not host's port is open.

    Parameters
    ----------
    host : str
        Host IP address or hostname
    port : int
        Port number

    Returns
    -------
    awaitable bool
    """
    try:
        _reader, writer = await asyncio.wait_for[
            asyncio.open_connection[host, port], timeout=2
        ]
        writer.close[]
        await writer.wait_closed[]
        return True
    except Exception:
        pass

    return False 
1

def open_port[host, port]:
    """Return whether or not host's port is open.

    Parameters
    ----------
    host : str
        Host IP address or hostname
    port : int
        Port number

    Returns
    -------
    awaitable bool
    """
    try:
        _reader, writer = await asyncio.wait_for[
            asyncio.open_connection[host, port], timeout=2
        ]
        writer.close[]
        await writer.wait_closed[]
        return True
    except Exception:
        pass

    return False 
2

def open_port[host, port]:
    """Return whether or not host's port is open.

    Parameters
    ----------
    host : str
        Host IP address or hostname
    port : int
        Port number

    Returns
    -------
    awaitable bool
    """
    try:
        _reader, writer = await asyncio.wait_for[
            asyncio.open_connection[host, port], timeout=2
        ]
        writer.close[]
        await writer.wait_closed[]
        return True
    except Exception:
        pass

    return False 
3

def open_port[host, port]:
    """Return whether or not host's port is open.

    Parameters
    ----------
    host : str
        Host IP address or hostname
    port : int
        Port number

    Returns
    -------
    awaitable bool
    """
    try:
        _reader, writer = await asyncio.wait_for[
            asyncio.open_connection[host, port], timeout=2
        ]
        writer.close[]
        await writer.wait_closed[]
        return True
    except Exception:
        pass

    return False 
4

def open_port[host, port]:
    """Return whether or not host's port is open.

    Parameters
    ----------
    host : str
        Host IP address or hostname
    port : int
        Port number

    Returns
    -------
    awaitable bool
    """
    try:
        _reader, writer = await asyncio.wait_for[
            asyncio.open_connection[host, port], timeout=2
        ]
        writer.close[]
        await writer.wait_closed[]
        return True
    except Exception:
        pass

    return False 
5

def open_port[host, port]:
    """Return whether or not host's port is open.

    Parameters
    ----------
    host : str
        Host IP address or hostname
    port : int
        Port number

    Returns
    -------
    awaitable bool
    """
    try:
        _reader, writer = await asyncio.wait_for[
            asyncio.open_connection[host, port], timeout=2
        ]
        writer.close[]
        await writer.wait_closed[]
        return True
    except Exception:
        pass

    return False 
6

def open_port[host, port]:
    """Return whether or not host's port is open.

    Parameters
    ----------
    host : str
        Host IP address or hostname
    port : int
        Port number

    Returns
    -------
    awaitable bool
    """
    try:
        _reader, writer = await asyncio.wait_for[
            asyncio.open_connection[host, port], timeout=2
        ]
        writer.close[]
        await writer.wait_closed[]
        return True
    except Exception:
        pass

    return False 
7

def open_port[host, port]:
    """Return whether or not host's port is open.

    Parameters
    ----------
    host : str
        Host IP address or hostname
    port : int
        Port number

    Returns
    -------
    awaitable bool
    """
    try:
        _reader, writer = await asyncio.wait_for[
            asyncio.open_connection[host, port], timeout=2
        ]
        writer.close[]
        await writer.wait_closed[]
        return True
    except Exception:
        pass

    return False 
8

def open_port[host, port]:
    """Return whether or not host's port is open.

    Parameters
    ----------
    host : str
        Host IP address or hostname
    port : int
        Port number

    Returns
    -------
    awaitable bool
    """
    try:
        _reader, writer = await asyncio.wait_for[
            asyncio.open_connection[host, port], timeout=2
        ]
        writer.close[]
        await writer.wait_closed[]
        return True
    except Exception:
        pass

    return False 
9

def check_port_open[port, loop, delay=1]:
    # the "s = socket.socket; s.bind" approach sometimes says a port is in use when it's not
    # this approach replicates aiohttp so should always give the same answer
    for i in range[5, 0, -1]:
        try:
            server = await loop.create_server[asyncio.Protocol[], host=HOST, port=port]
        except OSError as e:
            if e.errno != 98:  # pragma: no cover
                raise
            dft_logger.warning['port %d is already in use, waiting %d...', port, i]
            await asyncio.sleep[delay, loop=loop]
        else:
            server.close[]
            await server.wait_closed[]
            return
    raise AiohttpDevException['The port {} is already is use'.format[port]] 
0

def check_port_open[port, loop, delay=1]:
    # the "s = socket.socket; s.bind" approach sometimes says a port is in use when it's not
    # this approach replicates aiohttp so should always give the same answer
    for i in range[5, 0, -1]:
        try:
            server = await loop.create_server[asyncio.Protocol[], host=HOST, port=port]
        except OSError as e:
            if e.errno != 98:  # pragma: no cover
                raise
            dft_logger.warning['port %d is already in use, waiting %d...', port, i]
            await asyncio.sleep[delay, loop=loop]
        else:
            server.close[]
            await server.wait_closed[]
            return
    raise AiohttpDevException['The port {} is already is use'.format[port]] 
1

Đâu là sự khác biệt giữa bind[] và listen[] ?

Một máy chủ có một phương thức bind[] liên kết nó với một IP và cổng cụ thể để nó có thể lắng nghe các yêu cầu đến trên IP và cổng đó. Máy chủ có phương thức listen[] đưa máy chủ vào chế độ nghe. Điều này cho phép máy chủ lắng nghe các kết nối đến

Làm cách nào để kiểm tra các cổng đang mở trong Python?

Bạn có thể. .
Kiểm tra tất cả các cổng được sử dụng bởi một pid cụ thể. proc = psutil. Proc in quy trình [pid]. kết nối[]
Kiểm tra tất cả các cổng được sử dụng trên máy cục bộ. in psutil. net_connections[]

Làm cách nào để đọc dữ liệu từ cổng TCP trong Python?

Nó bắt đầu bằng cách tạo một ổ cắm TCP/IP. .
import socket import sys # Tạo socket TCP/IP sock = socket. ổ cắm [ổ cắm. .
# Liên kết ổ cắm với cổng server_address = ['localhost', 10000] print >>sys. stderr, 'khởi động trên %s cổng %s' % server_address sock. .
# Nghe các kết nối đến sock

Ổ cắm lắng nghe 5 Python là gì?

ổ cắm. lắng nghe [tồn đọng] Lắng nghe các kết nối được thực hiện với ổ cắm . Đối số tồn đọng chỉ định số lượng kết nối được xếp hàng đợi tối đa và ít nhất phải là 1; . Rõ ràng giá trị hệ thống là hơn 5 trên hệ thống của bạn.

Chủ Đề