crontabs.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145
  1. #
  2. # Copyright 2016, Martin Owens <doctormo@gmail.com>
  3. #
  4. # This library is free software; you can redistribute it and/or
  5. # modify it under the terms of the GNU Lesser General Public
  6. # License as published by the Free Software Foundation; either
  7. # version 3.0 of the License, or (at your option) any later version.
  8. #
  9. # This library is distributed in the hope that it will be useful,
  10. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  12. # Lesser General Public License for more details.
  13. #
  14. # You should have received a copy of the GNU Lesser General Public
  15. # License along with this library.
  16. #
  17. """
  18. The crontabs manager will list all available crontabs on the system.
  19. """
  20. import os
  21. import sys
  22. import pwd
  23. import itertools
  24. from os import stat, access, X_OK
  25. from pwd import getpwuid
  26. from crontab import CronTab
  27. class UserSpool(list):
  28. """Generates all user crontabs, yields both owned and abandoned tabs"""
  29. def __init__(self, loc, tabs=None):
  30. for username in self.listdir(loc):
  31. tab = self.generate(loc, username)
  32. if tab:
  33. self.append(tab)
  34. if not self:
  35. tab = CronTab(user=True)
  36. if tab:
  37. self.append(tab)
  38. def listdir(self, loc):
  39. try:
  40. return os.listdir(loc)
  41. except OSError:
  42. return []
  43. def get_owner(self, path):
  44. """Returns user file at path"""
  45. try:
  46. return getpwuid(stat(path).st_uid).pw_name
  47. except KeyError:
  48. return
  49. def generate(self, loc, username):
  50. path = os.path.join(loc, username)
  51. if username != self.get_owner(path):
  52. # Abandoned crontab pool entry!
  53. return CronTab(tabfile=path)
  54. return CronTab(user=username)
  55. class SystemTab(list):
  56. """Generates all system tabs"""
  57. def __init__(self, loc, tabs=None):
  58. if os.path.isdir(loc):
  59. for item in os.listdir(loc):
  60. if item[0] == '.':
  61. continue
  62. path = os.path.join(loc, item)
  63. self.append(CronTab(user=False, tabfile=path))
  64. elif os.path.isfile(loc):
  65. self.append(CronTab(user=False, tabfile=loc))
  66. class AnaCronTab(list):
  67. """Attempts to digest anacron entries (if possible)"""
  68. def __init__(self, loc, tabs=None):
  69. if tabs and os.path.isdir(loc):
  70. self.append(CronTab(user=False))
  71. jobs = list(tabs.all.find_command(loc))
  72. if jobs:
  73. for item in os.listdir(loc):
  74. self.add(loc, item, jobs[0])
  75. jobs[0].delete()
  76. def add(self, loc, item, anajob):
  77. path = os.path.join(loc, item)
  78. if item in ['0anacron'] or item[0] == '.' or not access(path, X_OK):
  79. return
  80. job = self[0].new(command=path, user=anajob.user)
  81. job.set_comment('Anacron %s' % loc.split('.')[-1])
  82. job.setall(anajob)
  83. return job
  84. # Files are direct, directories are listed (recursively)
  85. KNOWN_LOCATIONS = [
  86. # Known linux locations (Debian, RedHat, etc)
  87. (UserSpool, '/var/spool/cron/crontabs/'),
  88. (SystemTab, '/etc/crontab'),
  89. (SystemTab, '/etc/cron.d/'),
  90. # Anacron digestion (we want to know more)
  91. (AnaCronTab, '/etc/cron.hourly'),
  92. (AnaCronTab, '/etc/cron.daily'),
  93. (AnaCronTab, '/etc/cron.weekly'),
  94. (AnaCronTab, '/etc/cron.monthly'),
  95. # Known MacOSX locations
  96. # None
  97. # Other (windows, bsd)
  98. # None
  99. ]
  100. class CronTabs(list):
  101. """Singleton dictionary of all detectable crontabs"""
  102. _all = None
  103. _self = None
  104. def __new__(cls, *args, **kw):
  105. if not cls._self:
  106. cls._self = super(CronTabs, cls).__new__(cls, *args, **kw)
  107. return cls._self
  108. def __init__(self):
  109. if not self:
  110. for loc in KNOWN_LOCATIONS:
  111. self.add(*loc)
  112. def add(self, cls, *args):
  113. for tab in cls(*args, tabs=self):
  114. self.append(tab)
  115. self._all = None
  116. @property
  117. def all(self):
  118. """Return a CronTab object with all jobs (read-only)"""
  119. if self._all is None:
  120. self._all = CronTab(user=False)
  121. for tab in self:
  122. for job in tab:
  123. if job.user is None:
  124. job.user = tab.user or 'unknown'
  125. self._all.append(job)
  126. return self._all