Python socket.SO_PEERCRED属性代码示例

本文整理汇总了Python中socket.SO_PEERCRED属性的典型用法代码示例。如果您正苦于以下问题:Python socket.SO_PEERCRED属性的具体用法?Python socket.SO_PEERCRED怎么用?Python socket.SO_PEERCRED使用的例子?那么恭喜您, 这里精选的属性代码示例或许可以为您提供帮助。您也可以进一步了解该属性所在模块socket的用法示例。

在下文中一共展示了socket.SO_PEERCRED属性的5个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: lineReceived

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_PEERCRED [as 别名]
def lineReceived(self, line):
        """Process line from the clien."""
        creds = self.transport.socket.getsockopt(
            socket.SOL_SOCKET,
            socket.SO_PEERCRED,
            struct.calcsize('3i')
        )
        pid, uid, gid = struct.unpack('3i', creds)
        _LOGGER.info('Connection from pid: %d, uid: %d, gid %d', pid, uid, gid)
        try:
            self.username = utils.get_username(uid)
            self.uid = uid
            self.gid = gid
            decoded = base64.standard_b64decode(line)
            assert isinstance(decoded, bytes), repr(decoded)
            self.got_line(decoded)
        except KeyError:
            _LOGGER.warning('Unable to get username for uid: %d', uid)
            self.username = None
            self.transport.loseConnection() 
开发者ID:Morgan-Stanley,项目名称:treadmill,代码行数:25,代码来源:peercredprotocol.py


示例2: __call__

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_PEERCRED [as 别名]
def __call__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
                       ) -> None:
        sock = writer.transport.get_extra_info("socket")
        # TODO support non-linux OSes
        # I think FreeBSD uses SCM_CREDS
        creds = sock.getsockopt(SOL_SOCKET, SO_PEERCRED, struct.calcsize('3i'))
        pid, uid, gid = struct.unpack('3i', creds)
        user_info = pwd.getpwuid(uid)
        username = f"{user_info.pw_name} ({uid})" if user_info and user_info.pw_name else uid
        if len(self.whitelist) > 0 and uid not in self.whitelist:
            writer.write(b"You are not whitelisted to use the manhole.")
            log.warning(f"Non-whitelisted user {username} tried to connect from PID {pid}")
            await writer.drain()
            writer.close()
            return
        namespace = {**self.namespace}
        if AWAIT_TRANSFORM:
            namespace[AWAIT_FUNC_NAME] = AWAIT_FALLBACK
        interpreter = self.interpreter_class(namespace=namespace, banner=self.banner,
                                             loop=self.loop)
        namespace["exit"] = interpreter.close
        self.clients.append(interpreter)
        conn_id = self.conn_id
        log.info(f"Manhole connection OPENED: {conn_id} from PID {pid} by {username}")
        await asyncio.ensure_future(interpreter(reader, writer))
        log.info(f"Manhole connection CLOSED: {conn_id} from PID {pid} by {username}")
        self.clients.remove(interpreter) 
开发者ID:tulir,项目名称:mautrix-python,代码行数:31,代码来源:manhole.py


示例3: start_manhole

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_PEERCRED [as 别名]
def start_manhole(path: str, banner: str = "", namespace: Optional[Dict[str, Any]] = None,
                        loop: asyncio.AbstractEventLoop = None, whitelist: Set[int] = None,
                        ) -> Tuple[asyncio.AbstractServer, Callable[[], None]]:
    """
    Starts a manhole server on a given UNIX address.
    Args:
        path: The path to create the UNIX socket at.
        banner: The banner to show when clients connect.
        namespace: The globals to provide to connected clients.
        loop: The asyncio event loop to use.
        whitelist: List of user IDs to allow connecting.
    """
    if not SO_PEERCRED:
        raise ValueError("SO_PEERCRED is not supported on this platform")
    loop = loop or asyncio.get_event_loop()
    factory = InterpreterFactory(namespace=namespace, banner=banner,
                                 interpreter_class=AsyncInterpreter, loop=loop,
                                 whitelist=whitelist)
    server = await asyncio.start_unix_server(factory, path=path, loop=loop)
    os.chmod(path, 0o666)
    def stop():
        for client in factory.clients:
            client.close()
        server.close()
    return server, stop 
开发者ID:tulir,项目名称:mautrix-python,代码行数:30,代码来源:manhole.py


示例4: accept

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_PEERCRED [as 别名]
def accept(self, sock):
        loop = get_event_loop()
        accept_sock, _ = await loop.sock_accept(sock)
        pid = accept_sock.getsockopt(SOL_SOCKET, SO_PEERCRED)
        write_text_file(path.join(self.cpuacct_cgroup_dir, 'tasks'), str(pid))
        write_text_file(path.join(self.memory_cgroup_dir, 'tasks'), str(pid))
        write_text_file(path.join(self.pids_cgroup_dir, 'tasks'), str(pid))
        accept_sock.close() 
开发者ID:vijos,项目名称:jd4,代码行数:10,代码来源:cgroup.py


示例5: get_peer_creds

# 需要导入模块: import socket [as 别名]
# 或者: from socket import SO_PEERCRED [as 别名]
def get_peer_creds(self):  # LRU cached on per-instance basis, see __init__
        """Return the PID/UID/GID tuple of the peer socket for UNIX sockets.
        This function uses SO_PEERCRED to query the UNIX PID, UID, GID
        of the peer, which is only available if the bind address is
        a UNIX domain socket.
        Raises:
            NotImplementedError: in case of unsupported socket type
            RuntimeError: in case of SO_PEERCRED lookup unsupported or disabled
        """
        PEERCRED_STRUCT_DEF = '3i'
        if IS_WINDOWS or self.socket.family != socket.AF_UNIX:
            raise NotImplementedError(
                'SO_PEERCRED is only supported in Linux kernel and WSL',
            )
        elif not self.peercreds_enabled:
            raise RuntimeError(
                'Peer creds lookup is disabled within this server',
            )
        try:
            peer_creds = self.socket.getsockopt(
                # FIXME: Use LOCAL_CREDS for BSD-like OSs
                # Ref: https://gist.github.com/LucaFilipozzi/e4f1e118202aff27af6aadebda1b5d91  # noqa
                socket.SOL_SOCKET, socket.SO_PEERCRED,
                struct.calcsize(PEERCRED_STRUCT_DEF),
            )
        except socket.error as socket_err:
            """Non-Linux kernels don't support SO_PEERCRED.
            Refs:
            http://welz.org.za/notes/on-peer-cred.html
            https://github.com/daveti/tcpSockHack
            msdn.microsoft.com/en-us/commandline/wsl/release_notes#build-15025
            """
            six.raise_from(  # 3.6+: raise RuntimeError from socket_err
                RuntimeError,
                socket_err,
            )
        else:
            pid, uid, gid = struct.unpack(PEERCRED_STRUCT_DEF, peer_creds)
            return pid, uid, gid 
开发者ID:cherrypy,项目名称:cheroot,代码行数:47,代码来源:server.py



注:本文中的socket.SO_PEERCRED属性示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。