pdm/srv.py

   1 """Python Daemon Management -- Server functions
   2
   3 This module implements the server part of the PDM protocols. The
   4 primary object of interest herein is the listen() function, which is
   5 the most generic way to create PDM listeners based on user
   6 configuration, and the documentation for the repl and perf classes,
   7 which describes the functioning of the REPL and PERF protocols.
   8 """
   9
  10 import os, sys, socket, threading, grp, select
  11 import types, pprint, traceback
  12 import pickle, struct
  13
  14 __all__ = ["repl", "perf", "listener", "unixlistener", "tcplistener", "listen"]
  15
  16 protocols = {}
  17
  18 class repl(object):
  19     """REPL protocol handler
  20
  21     Provides a read-eval-print loop. The primary client-side interface
  22     is the L{pdm.cli.replclient} class. Clients can send arbitrary
  23     code, which is compiled and run on its own thread in the server
  24     process, and output responses that are echoed back to the client.
  25
  26     Each client is provided with its own module, in which the code
  27     runs. The module is prepared with a function named `echo', which
  28     takes a single object and pretty-prints it as part of the command
  29     response. If a command can be parsed as an expression, the value
  30     it evaluates to is automatically echoed to the client. If the
  31     evalution of the command terminates with an exception, its
  32     traceback is echoed to the client.
  33
  34     The REPL protocol is only intended for interactive usage. In order
  35     to interact programmatically with the server process, see the PERF
  36     protocol instead.
  37     """
  38     def __init__(self, cl):
  39         self.cl = cl
  40         self.mod = types.ModuleType("repl")
  41         self.mod.echo = self.echo
  42         self.printer = pprint.PrettyPrinter(indent = 4, depth = 6)
  43         cl.send(b"+REPL\n")
  44
  45     def sendlines(self, text):
  46         for line in text.split("\n"):
  47             self.cl.send(b" " + line.encode("utf-8") + b"\n")
  48
  49     def echo(self, ob):
  50         self.sendlines(self.printer.pformat(ob))
  51
  52     def command(self, cmd):
  53         cmd = cmd.decode("utf-8")
  54         try:
  55             try:
  56                 ccode = compile(cmd, "PDM Input", "eval")
  57             except SyntaxError:
  58                 ccode = compile(cmd, "PDM Input", "exec")
  59                 exec(ccode, self.mod.__dict__)
  60                 self.cl.send(b"+OK\n")
  61             else:
  62                 self.echo(eval(ccode, self.mod.__dict__))
  63                 self.cl.send(b"+OK\n")
  64         except:
  65             lines = ("".join(traceback.format_exception(*sys.exc_info()))).split("\n")
  66             while len(lines) > 0 and lines[-1] == "": lines = lines[:-1]
  67             for line in lines:
  68                 self.cl.send(b" " + line.encode("utf-8") + b"\n")
  69             self.cl.send(b"+EXC\n")
  70
  71     def handle(self, buf):
  72         p = buf.find(b"\n\n")
  73         if p < 0:
  74             return buf
  75         cmd = buf[:p + 1]
  76         self.command(cmd)
  77         return buf[p + 2:]
  78 protocols["repl"] = repl
  79
  80 class perf(object):
  81     """PERF protocol handler
  82
  83     The PERF protocol provides an interface for program interaction
  84     with the server process. It allows limited remote interactions
  85     with Python objects over a few defined interfaces.
  86
  87     All objects that wish to be available for interaction need to
  88     implement a method named `pdm_protocols' which, when called with
  89     no arguments, should return a list of strings, each indicating a
  90     PERF interface that the object implements. For each such
  91     interface, the object must implement additional methods as
  92     described below.
  93
  94     A client can find PERF objects to interact with either by
  95     specifying the name of such an object in an existing module, or by
  96     using the `dir' interface, described below. Thus, to make a PERF
  97     object available for clients, it needs only be bound to a global
  98     variable in a module and implement the `pdm_protocols'
  99     method. When requesting an object from a module, the module must
 100     already be imported. PDM will not import new modules for clients;
 101     rather, the daemon process needs to import all modules that
 102     clients should be able to interact with. PDM itself always imports
 103     the L{pdm.perf} module, which contains a few basic PERF
 104     objects. See its documentation for details.
 105
 106     The following interfaces are currently known to PERF.
 107
 108      - attr:
 109        An object that implements the `attr' interface models an
 110        attribute that can be read by clients. The attribute can be
 111        anything, as long as its representation can be
 112        pickled. Examples of attributes could be such things as the CPU
 113        time consumed by the server process, or the number of active
 114        connections to whatever clients the program serves. To
 115        implement the `attr' interface, an object must implement
 116        methods called `readattr' and `attrinfo'. `readattr' is called
 117        with no arguments to read the current value of the attribute,
 118        and `attrinfo' is called with no arguments to read a
 119        description of the attribute. Both should be
 120        idempotent. `readattr' can return any pickleable object, and
 121        `attrinfo' should return either None to indicate that it has no
 122        description, or an instance of the L{pdm.perf.attrinfo} class.
 123
 124      - dir:
 125        The `dir' interface models a directory of other PERF
 126        objects. An object implementing it must implement methods
 127        called `lookup' and `listdir'. `lookup' is called with a single
 128        string argument that names an object, and should either return
 129        another PERF object based on the name, or raise KeyError if it
 130        does not recognize the name. `listdir' is called with no
 131        arguments, and should return a list of known names that can be
 132        used as argument to `lookup', but the list is not required to
 133        be exhaustive and may also be empty.
 134
 135      - invoke:
 136        The `invoke' interface allows a more arbitrary form of method
 137        calls to objects implementing it. Such objects must implement a
 138        method called `invoke', which is called with one positional
 139        argument naming a method to be called (which it is free to
 140        interpret however it wishes), and with any additional
 141        positional and keyword arguments that the client wishes to pass
 142        to it. Whatever `invoke' returns is pickled and sent back to
 143        the client. In case the method name is not recognized, `invoke'
 144        should raise an AttributeError.
 145
 146      - event:
 147        The `event' interface allows PERF objects to notify clients of
 148        events asynchronously. Objects implementing it must implement
 149        methods called `subscribe' and `unsubscribe'. `subscribe' will
 150        be called with a single argument, which is a callable of one
 151        argument, which should be registered to be called when an event
 152        pertaining to the `event' object in question occurs. The
 153        `event' object should then call all such registered callables
 154        with a single argument describing the event. The argument could
 155        be any object that can be pickled, but should be an instance of
 156        a subclass of the L{pdm.perf.event} class. If `subscribe' is
 157        called with a callback object that it has already registered,
 158        it should raise a ValueError. `unsubscribe' is called with a
 159        single argument, which is a previously registered callback
 160        object, which should then be unregistered to that it is no
 161        longer called when an event occurs. If the given callback
 162        object is not, in fact, registered, a ValueError should be
 163        raised.
 164
 165     The L{pdm.perf} module contains a few convenience classes which
 166     implements the interfaces, but PERF objects are not required to be
 167     instances of them. Any object can implement a PERF interface, as
 168     long as it does so as described above.
 169
 170     The L{pdm.cli.perfclient} class is the client-side implementation.
 171     """
 172     def __init__(self, cl):
 173         self.cl = cl
 174         self.odtab = {}
 175         cl.send(b"+PERF1\n")
 176         self.buf = ""
 177         self.lock = threading.Lock()
 178         self.subscribed = {}
 179
 180     def closed(self):
 181         for id, recv in self.subscribed.items():
 182             ob = self.odtab[id]
 183             if ob is None: continue
 184             ob, protos = ob
 185             try:
 186                 ob.unsubscribe(recv)
 187             except: pass
 188
 189     def send(self, *args):
 190         self.lock.acquire()
 191         try:
 192             buf = pickle.dumps(args)
 193             buf = struct.pack(">l", len(buf)) + buf
 194             self.cl.send(buf)
 195         finally:
 196             self.lock.release()
 197
 198     def bindob(self, id, ob):
 199         if not hasattr(ob, "pdm_protocols"):
 200             raise ValueError("Object does not support PDM introspection")
 201         try:
 202             proto = ob.pdm_protocols()
 203         except Exception as exc:
 204             raise ValueError("PDM introspection failed", exc)
 205         self.odtab[id] = ob, proto
 206         return proto
 207
 208     def bind(self, id, module, obnm):
 209         resmod = sys.modules.get(module)
 210         if resmod is None:
 211             self.send("-", ImportError("No such module: %s" % module))
 212             return
 213         try:
 214             ob = getattr(resmod, obnm)
 215         except AttributeError:
 216             self.send("-", AttributeError("No such object: %s" % obnm))
 217             return
 218         try:
 219             proto = self.bindob(id, ob)
 220         except Exception as exc:
 221             self.send("-", exc)
 222             return
 223         self.send("+", proto)
 224
 225     def getob(self, id, proto):
 226         ob = self.odtab.get(id)
 227         if ob is None:
 228             self.send("-", ValueError("No such bound ID: %r" % id))
 229             return None
 230         ob, protos = ob
 231         if proto not in protos:
 232             self.send("-", ValueError("Object does not support that protocol"))
 233             return None
 234         return ob
 235
 236     def lookup(self, tgtid, srcid, obnm):
 237         src = self.getob(srcid, "dir")
 238         if src is None:
 239             return
 240         try:
 241             ob = src.lookup(obnm)
 242         except KeyError as exc:
 243             self.send("-", exc)
 244             return
 245         try:
 246             proto = self.bindob(tgtid, ob)
 247         except Exception as exc:
 248             self.send("-", exc)
 249             return
 250         self.send("+", proto)
 251
 252     def unbind(self, id):
 253         ob = self.odtab.get(id)
 254         if ob is None:
 255             self.send("-", KeyError("No such name bound: %r" % id))
 256             return
 257         ob, protos = ob
 258         del self.odtab[id]
 259         recv = self.subscribed.get(id)
 260         if recv is not None:
 261             ob.unsubscribe(recv)
 262             del self.subscribed[id]
 263         self.send("+")
 264
 265     def listdir(self, id):
 266         ob = self.getob(id, "dir")
 267         if ob is None:
 268             return
 269         self.send("+", ob.listdir())
 270
 271     def readattr(self, id):
 272         ob = self.getob(id, "attr")
 273         if ob is None:
 274             return
 275         try:
 276             ret = ob.readattr()
 277         except Exception as exc:
 278             self.send("-", Exception("Could not read attribute"))
 279             return
 280         self.send("+", ret)
 281
 282     def attrinfo(self, id):
 283         ob = self.getob(id, "attr")
 284         if ob is None:
 285             return
 286         self.send("+", ob.attrinfo())
 287
 288     def invoke(self, id, method, args, kwargs):
 289         ob = self.getob(id, "invoke")
 290         if ob is None:
 291             return
 292         try:
 293             self.send("+", ob.invoke(method, *args, **kwargs))
 294         except Exception as exc:
 295             self.send("-", exc)
 296
 297     def event(self, id, ob, ev):
 298         self.send("*", id, ev)
 299
 300     def subscribe(self, id):
 301         ob = self.getob(id, "event")
 302         if ob is None:
 303             return
 304         if id in self.subscribed:
 305             self.send("-", ValueError("Already subscribed"))
 306         def recv(ev):
 307             self.event(id, ob, ev)
 308         ob.subscribe(recv)
 309         self.subscribed[id] = recv
 310         self.send("+")
 311
 312     def unsubscribe(self, id):
 313         ob = self.getob(id, "event")
 314         if ob is None:
 315             return
 316         recv = self.subscribed.get(id)
 317         if recv is None:
 318             self.send("-", ValueError("Not subscribed"))
 319         ob.unsubscribe(recv)
 320         del self.subscribed[id]
 321         self.send("+")
 322
 323     def command(self, data):
 324         cmd = data[0]
 325         if cmd == "bind":
 326             self.bind(*data[1:])
 327         elif cmd == "unbind":
 328             self.unbind(*data[1:])
 329         elif cmd == "lookup":
 330             self.lookup(*data[1:])
 331         elif cmd == "ls":
 332             self.listdir(*data[1:])
 333         elif cmd == "readattr":
 334             self.readattr(*data[1:])
 335         elif cmd == "attrinfo":
 336             self.attrinfo(*data[1:])
 337         elif cmd == "invoke":
 338             self.invoke(*data[1:])
 339         elif cmd == "subs":
 340             self.subscribe(*data[1:])
 341         elif cmd == "unsubs":
 342             self.unsubscribe(*data[1:])
 343         else:
 344             self.send("-", Exception("Unknown command: %r" % (cmd,)))
 345
 346     def handle(self, buf):
 347         if len(buf) < 4:
 348             return buf
 349         dlen = struct.unpack(">l", buf[:4])[0]
 350         if len(buf) < dlen + 4:
 351             return buf
 352         data = pickle.loads(buf[4:dlen + 4])
 353         self.command(data)
 354         return buf[dlen + 4:]
 355
 356 protocols["perf"] = perf
 357
 358 class client(threading.Thread):
 359     def __init__(self, sk):
 360         super().__init__(name = "Management client")
 361         self.setDaemon(True)
 362         self.sk = sk
 363         self.handler = self
 364
 365     def send(self, data):
 366         return self.sk.send(data)
 367
 368     def choose(self, proto):
 369         try:
 370             proto = proto.decode("ascii")
 371         except UnicodeError:
 372             proto = None
 373         if proto in protocols:
 374             self.handler = protocols[proto](self)
 375         else:
 376             self.send("-ERR Unknown protocol: %s\n" % proto)
 377             raise Exception()
 378
 379     def handle(self, buf):
 380         p = buf.find(b"\n")
 381         if p >= 0:
 382             proto = buf[:p]
 383             buf = buf[p + 1:]
 384             self.choose(proto)
 385         return buf
 386
 387     def run(self):
 388         try:
 389             buf = b""
 390             self.send(b"+PDM1\n")
 391             while True:
 392                 ret = self.sk.recv(1024)
 393                 if ret == b"":
 394                     return
 395                 buf += ret
 396                 while True:
 397                     try:
 398                         nbuf = self.handler.handle(buf)
 399                     except:
 400                         #for line in traceback.format_exception(*sys.exc_info()):
 401                         #    print(line)
 402                         return
 403                     if nbuf == buf:
 404                         break
 405                     buf = nbuf
 406         finally:
 407             try:
 408                 self.sk.close()
 409             finally:
 410                 if hasattr(self.handler, "closed"):
 411                     self.handler.closed()
 412
 413
 414 class listener(threading.Thread):
 415     """PDM listener
 416
 417     This subclass of a thread listens to PDM connections and handles
 418     client connections properly. It is intended to be subclassed by
 419     providers of specific domains, such as unixlistener and
 420     tcplistener.
 421     """
 422     def __init__(self):
 423         super().__init__(name = "Management listener")
 424         self.setDaemon(True)
 425
 426     def listen(self, sk):
 427         """Listen for and accept connections."""
 428         self.running = True
 429         while self.running:
 430             rfd, wfd, efd = select.select([sk], [], [sk], 1)
 431             for fd in rfd:
 432                 if fd == sk:
 433                     nsk, addr = sk.accept()
 434                     self.accept(nsk, addr)
 435
 436     def stop(self):
 437         """Stop listening for client connections
 438
 439         Tells the listener thread to stop listening, and then waits
 440         for it to terminate.
 441         """
 442         self.running = False
 443         self.join()
 444
 445     def accept(self, sk, addr):
 446         cl = client(sk)
 447         cl.start()
 448
 449 class unixlistener(listener):
 450     """Unix socket listener"""
 451     def __init__(self, name, mode = 0o600, group = None):
 452         """Create a listener that will bind to the Unix socket named
 453         by `name'. The socket will not actually be bound until the
 454         listener is started. The socket will be chmodded to `mode',
 455         and if `group' is given, the named group will be set as the
 456         owner of the socket.
 457         """
 458         super().__init__()
 459         self.name = name
 460         self.mode = mode
 461         self.group = group
 462
 463     def run(self):
 464         sk = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
 465         ul = False
 466         try:
 467             if os.path.exists(self.name) and os.path.stat.S_ISSOCK(os.stat(self.name).st_mode):
 468                 os.unlink(self.name)
 469             sk.bind(self.name)
 470             ul = True
 471             os.chmod(self.name, self.mode)
 472             if self.group is not None:
 473                 os.chown(self.name, os.getuid(), grp.getgrnam(self.group).gr_gid)
 474             sk.listen(16)
 475             self.listen(sk)
 476         finally:
 477             sk.close()
 478             if ul:
 479                 os.unlink(self.name)
 480
 481 class tcplistener(listener):
 482     """TCP socket listener"""
 483     def __init__(self, port, bindaddr = "127.0.0.1"):
 484         """Create a listener that will bind to the given TCP port, and
 485         the given local interface. The socket will not actually be
 486         bound until the listener is started.
 487         """
 488         super().__init__()
 489         self.port = port
 490         self.bindaddr = bindaddr
 491
 492     def run(self):
 493         sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 494         try:
 495             sk.bind((self.bindaddr, self.port))
 496             sk.listen(16)
 497             self.listen(sk)
 498         finally:
 499             sk.close()
 500
 501 def listen(spec):
 502     """Create and start a listener according to a string
 503     specification. The string specifications can easily be passed from
 504     command-line options, user configuration or the like. Currently,
 505     the two following specification formats are recognized:
 506
 507     PATH[:MODE[:GROUP]] -- PATH must contain at least one slash. A
 508     Unix socket listener will be created listening to that path, and
 509     the socket will be chmodded to MODE and owned by GROUP. If MODE is
 510     not given, it defaults to 0600, and if GROUP is not given, the
 511     process' default group is used.
 512
 513     ADDRESS:PORT -- PORT must be entirely numeric. A TCP socket
 514     listener will be created listening to that port, bound to the
 515     given local interface address. Since PDM has no authentication
 516     support, ADDRESS should probably be localhost.
 517     """
 518     if ":" in spec:
 519         first = spec[:spec.index(":")]
 520         last = spec[spec.rindex(":") + 1:]
 521     else:
 522         first = spec
 523         last = spec
 524     if "/" in first:
 525         parts = spec.split(":")
 526         mode = 0o600
 527         group = None
 528         if len(parts) > 1:
 529             mode = int(parts[1], 8)
 530         if len(parts) > 2:
 531             group = parts[2]
 532         ret = unixlistener(parts[0], mode = mode, group = group)
 533         ret.start()
 534         return ret
 535     if last.isdigit():
 536         p = spec.rindex(":")
 537         host = spec[:p]
 538         port = int(spec[p + 1:])
 539         ret = tcplistener(port, bindaddr = host)
 540         ret.start()
 541         return ret
 542     raise ValueError("Unparsable listener specification: %r" % spec)
 543
 544 import pdm.perf