Make crontab configurable

restic_compose_backup.env

@@ -10,6 +10,7 @@ RESTIC_KEEP_MONTHLY=12
 RESTIC_KEEP_YEARLY=3
 
 LOG_LEVEL=info
+CRON_SCHEDULE=10 2 * * *
 
 # EMAIL_HOST=
 # EMAIL_PORT=

src/crontab

@@ -1 +1,2 @@
-0 2 * * * source /env.sh && rcb backup > /proc/1/fd/1
+10 2 * * * source /env.sh && rcb backup > /proc/1/fd/1
+

src/entrypoint.sh

@@ -3,6 +3,9 @@
 # Dump all env vars so we can source them in cron jobs
 printenv | sed 's/^\(.*\)$/export \1/g' > /env.sh
 
+# Write crontab
+rcb crontab > crontab
+
 # start cron in the foreground
 crontab crontab
 crond -f

src/restic_compose_backup/cli.py

@@ -10,7 +10,7 @@ from restic_compose_backup import (
 )
 from restic_compose_backup.config import Config
 from restic_compose_backup.containers import RunningContainers
-from restic_compose_backup import utils
+from restic_compose_backup import cron, utils
 
 logger = logging.getLogger(__name__)
 
@@ -48,6 +48,9 @@ def main():
         import restic_compose_backup
         print(restic_compose_backup.__version__)
 
+    elif args.action == "crontab":
+        crontab(config)
+
 
 def status(config, containers):
     """Outputs the backup config for the compose setup"""
@@ -252,11 +255,25 @@ def alert(config, containers):
     )
 
 
+def crontab(config):
+    """Generate the crontab"""
+    print(cron.generate_crontab(config))
+
+
 def parse_args():
     parser = argparse.ArgumentParser(prog='restic_compose_backup')
     parser.add_argument(
         'action',
-        choices=['status', 'snapshots', 'backup', 'start-backup-process', 'alert', 'cleanup', 'version'],
+        choices=[
+            'status',
+            'snapshots',
+            'backup',
+            'start-backup-process',
+            'alert',
+            'cleanup',
+            'version',
+            'crontab',
+        ],
     )
     parser.add_argument(
         '--log-level',

src/restic_compose_backup/config.py

@@ -2,12 +2,17 @@ import os
 
 
 class Config:
+    default_backup_command = "source /env.sh && rcb backup > /proc/1/fd/1"
+    default_crontab_schedule = "0 2 * * *"
+
     """Bag for config values"""
     def __init__(self, check=True):
         # Mandatory values
         self.repository = os.environ.get('RESTIC_REPOSITORY')
         self.password = os.environ.get('RESTIC_REPOSITORY')
         self.docker_base_url = os.environ.get('DOCKER_BASE_URL') or "unix://tmp/docker.sock"
+        self.cron_schedule = os.environ.get('CRON_SCHEDULE') or self.default_crontab_schedule
+        self.cron_command = os.environ.get('CRON_COMMAND') or self.default_backup_command
 
         # Log
         self.log_level = os.environ.get('LOG_LEVEL')

src/restic_compose_backup/cron.py

@@ -0,0 +1,69 @@
+"""
+# ┌───────────── minute (0 - 59)
+# │ ┌───────────── hour (0 - 23)
+# │ │ ┌───────────── day of the month (1 - 31)
+# │ │ │ ┌───────────── month (1 - 12)
+# │ │ │ │ ┌───────────── day of the week (0 - 6) (Sunday to Saturday;
+# │ │ │ │ │                                   7 is also Sunday on some systems)
+# │ │ │ │ │
+# │ │ │ │ │
+# * * * * * command to execute
+"""
+QUOTE_CHARS = ['"', "'"]
+
+
+def generate_crontab(config):
+    """Generate a crontab entry for running backup job"""
+    command = config.cron_command.strip()
+    schedule = config.cron_schedule
+
+    if schedule:
+        schedule = schedule.strip()
+        schedule = strip_quotes(schedule)
+        if not validate_schedule(schedule):
+            schedule = config.default_crontab_schedule
+    else:
+        schedule = config.default_crontab_schedule
+
+    return f'{schedule} {command}\n'
+
+
+def validate_schedule(schedule: str):
+    """Validate crontab format"""
+    parts = schedule.split()
+    if len(parts) != 5:
+        return False
+
+    for p in parts:
+        if p != '*' and not p.isdigit():
+            return False
+
+    minute, hour, day, month, weekday = parts
+    try:
+        validate_field(minute, 0, 59)
+        validate_field(hour, 0, 23)
+        validate_field(day, 1, 31)
+        validate_field(month, 1, 12)
+        validate_field(weekday, 0, 6)
+    except ValueError:
+        return False
+
+    return True
+
+
+def validate_field(value, min, max):
+    if value == '*':
+        return
+
+    i = int(value)
+    return min <= i <= max
+
+
+def strip_quotes(value: str):
+    """Strip enclosing single or double quotes if present"""
+    if value[0] in QUOTE_CHARS:
+        value = value[1:]
+    if value[-1] in QUOTE_CHARS:
+        value = value[:-1]
+
+    return value