|
6 | 6 | # |
7 | 7 | # Based on https://github.com/nhoad/aiomanhole Copyright (c) 2014, Nathan Hoad |
8 | 8 | from typing import Any, Tuple, Optional, Dict, Union, List, Set, Callable, Type |
9 | | -from socket import SOL_SOCKET, SO_PEERCRED |
| 9 | +from socket import SOL_SOCKET |
10 | 10 | from abc import ABC, abstractmethod |
11 | 11 | from io import BytesIO, StringIO |
12 | 12 | from types import CodeType |
|
22 | 22 | import sys |
23 | 23 | import os |
24 | 24 |
|
| 25 | +try: |
| 26 | + from socket import SO_PEERCRED |
| 27 | +except ImportError: |
| 28 | + SO_PEERCRED = None |
| 29 | + |
25 | 30 | log = logging.getLogger("mau.manhole") |
26 | 31 |
|
27 | 32 |
|
@@ -301,6 +306,8 @@ def conn_id(self) -> int: |
301 | 306 | async def __call__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter |
302 | 307 | ) -> None: |
303 | 308 | sock = writer.transport.get_extra_info("socket") |
| 309 | + # TODO support non-linux OSes |
| 310 | + # I think FreeBSD uses SCM_CREDS |
304 | 311 | creds = sock.getsockopt(SOL_SOCKET, SO_PEERCRED, struct.calcsize('3i')) |
305 | 312 | pid, uid, gid = struct.unpack('3i', creds) |
306 | 313 | user_info = pwd.getpwuid(uid) |
@@ -340,6 +347,8 @@ async def start_manhole(path: str, banner: str = "", namespace: Optional[Dict[st |
340 | 347 | loop: The asyncio event loop to use. |
341 | 348 | whitelist: List of user IDs to allow connecting. |
342 | 349 | """ |
| 350 | + if not SO_PEERCRED: |
| 351 | + raise ValueError("SO_PEERCRED is not supported on this platform") |
343 | 352 | loop = loop or asyncio.get_event_loop() |
344 | 353 | factory = InterpreterFactory(namespace=namespace, banner=banner, |
345 | 354 | interpreter_class=AsyncInterpreter, loop=loop, |
|
0 commit comments