python/rox/su.py

   1 """The su (switch user) module allows you to execute commands as root.
   2 Typical usage:
   3
   4 def fn():
   5         proxy_maker = SuProxyMaker('I need root access to change /etc/fstab')
   6         yield proxy_maker.blocker
   7         root = proxy_maker.get_root()
   8
   9         call = root.open('/etc/fstab')
  10         yield call
  11         fd = call.result
  12
  13         ...
  14
  15         root.finish()
  16
  17 tasks.run(fn())
  18 """
  19
  20 import os, sys
  21 import rox
  22 from rox import g, _, master_proxy, tasks
  23 import traceback
  24 import fcntl
  25
  26 _my_dir = os.path.abspath(os.path.dirname(__file__))
  27 _child_script = os.path.join(_my_dir, 'suchild.sh')
  28
  29 class SuProxyMaker(tasks.Blocker):
  30         blocker = None
  31
  32         def __init__(self, message):
  33                 """Displays a box prompting the user for a password.
  34                 Creates a new master_proxy.MasterObject and starts the child
  35                 process. The user is prompted for the root
  36                 password."""
  37                 to_child = _Pipe()
  38                 from_child = _Pipe()
  39
  40                 if os.fork() == 0:
  41                         try:
  42                                 try:
  43                                         to_child.writeable.close()
  44                                         from_child.readable.close()
  45                                         _exec_child(message,
  46                                                     from_child.writeable,
  47                                                     to_child.readable)
  48                                 except:
  49                                         traceback.print_exc()
  50                         finally:
  51                                 os._exit(1)
  52                 from_child.writeable.close()
  53                 to_child.readable.close()
  54
  55                 self._root = master_proxy.MasterProxy(to_child.writeable,
  56                                                       from_child.readable).root
  57
  58                 self.blocker = self._root.getuid()
  59
  60         def get_root(self):
  61                 """Raises UserAbort if the user cancels."""
  62                 try:
  63                         uid = self.blocker.result
  64                 except master_proxy.LostConnection:
  65                         raise rox.UserAbort("Failed to become root (cancelled "
  66                                 "at user's request?)")
  67                 assert uid == 0
  68                 self.blocker = None
  69                 return self._root
  70
  71 def _exec_child(message, to_parent, from_parent):
  72         fcntl.fcntl(to_parent, fcntl.F_SETFD, 0)
  73         fcntl.fcntl(from_parent, fcntl.F_SETFD, 0)
  74         os.execlp('xterm', 'xterm',
  75                 '-geometry', '40x10',
  76                 '-title', 'Enter password',
  77                 '-e',
  78                 _child_script,
  79                 message,
  80                 sys.executable,
  81                 str(to_parent.fileno()),
  82                 str(from_parent.fileno()))
  83
  84 class _Pipe:
  85         """Contains Python file objects for two pipe ends.
  86         Wrapping the FDs in this way ensures that they will
  87         be freed on error."""
  88         readable = None
  89         writeable = None
  90
  91         def __init__(self):
  92                 r, w = os.pipe()
  93                 try:
  94                         self.readable = os.fdopen(r, 'r')
  95                 except:
  96                         os.close(r)
  97                         raise
  98                 try:
  99                         self.writeable = os.fdopen(w, 'w')
 100                 except:
 101                         os.close(w)
 102                         raise
 103