mirror of
https://github.com/sshuttle/sshuttle.git
synced 2024-11-28 10:53:21 +01:00
3bfb975ed9
* re-organized imports according to pep8 * fixed all remaining pep8 issues * moved common config into setup.cfg, additionally test `tests` * removed --select=X -- the errors selected where by default not in flake8's --ignore list so effectively had no effect * update .travis.yml to reflect changes in tox.ini * make travis just use tox in order to avoid code duplaction * replace py.test with pytest * fixed .travis.yml * try different pypy toxenv * hopefully fixed testenv for pypy * added pypy basepython, removed unused python2.6 * install dev package before testing (fixes missing coverage) * fixed empty exception pass blocks with noqa * Added dummy log message on empty try-except-pass blocks to make dodacy happy :( * Replaced Exception with BaseException
47 lines
935 B
Python
47 lines
935 B
Python
import socket
|
|
import os
|
|
|
|
from sshuttle.helpers import debug1
|
|
|
|
|
|
def _notify(message):
|
|
addr = os.environ.get("NOTIFY_SOCKET", None)
|
|
|
|
if not addr or len(addr) == 1 or addr[0] not in ('/', '@'):
|
|
return False
|
|
|
|
addr = '\0' + addr[1:] if addr[0] == '@' else addr
|
|
|
|
try:
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
|
|
except (OSError, IOError) as e:
|
|
debug1("Error creating socket to notify systemd: %s\n" % e)
|
|
return False
|
|
|
|
if not message:
|
|
return False
|
|
|
|
assert isinstance(message, bytes)
|
|
|
|
try:
|
|
return (sock.sendto(message, addr) > 0)
|
|
except (OSError, IOError) as e:
|
|
debug1("Error notifying systemd: %s\n" % e)
|
|
return False
|
|
|
|
|
|
def send(*messages):
|
|
return _notify(b'\n'.join(messages))
|
|
|
|
|
|
def ready():
|
|
return b"READY=1"
|
|
|
|
|
|
def stop():
|
|
return b"STOPPING=1"
|
|
|
|
|
|
def status(message):
|
|
return b"STATUS=%s" % message.encode('utf8')
|