Replaced cron with python scheduler and using python yaml library. Updated readme.

This commit is contained in:
Djeex 2025-06-01 10:44:06 +00:00
parent 7ead6b4154
commit 1236970426
6 changed files with 201 additions and 255 deletions

View File

@ -1,30 +1,17 @@
FROM python:3.11-slim
# Install required utilities
RUN apt-get update && apt-get install -y \
curl \
cron \
tzdata \
&& rm -rf /var/lib/apt/lists/*
RUN apt-get update && apt-get install -y --no-install-recommends curl tzdata && rm -rf /var/lib/apt/lists/*
# Install python dependencies
RUN pip install --no-cache-dir requests
RUN pip install --no-cache-dir requests pyyaml schedule
# Create crontabs directory (if needed)
RUN mkdir -p /etc/crontabs
ENV TZ=Europe/Paris
# Copy scripts
COPY update-blocklist.py /usr/local/bin/update-blocklist.py
COPY entrypoint.py /usr/local/bin/entrypoint.py
# Make scripts executable
RUN chmod +x /usr/local/bin/update-blocklist.py /usr/local/bin/entrypoint.py
# Set default timezone (can be overridden with TZ env var)
ENV TZ=UTC
# Configure timezone (tzdata)
RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
# Set entrypoint
ENTRYPOINT ["/usr/local/bin/entrypoint.py"]
WORKDIR /app
COPY blocklist_scheduler.py /app/blocklist_scheduler.py
RUN chmod +x /app/blocklist_scheduler.py
ENTRYPOINT ["python3", "/app/blocklist_scheduler.py"]

View File

@ -13,28 +13,37 @@
- [Features](#features)
- [Environment Variables](#environment-variables)
- [Volumes](#volumes)
- [File Structure](#file-structure)
- [Installation and Usage](#nstallation-and-usage)
- [Installation and Usage](#installation-and-usage)
## Features
- Automatically downloads IP CIDR blocks for specified countries to block.
- Supports additional manually blocked IPs from a configurable file.
- Updates the disallowed_clients section in the AdGuard Home config.
- Configurable update frequency via cron expression environment variable.
- Automatically restarts the AdGuard Home container after updates via Docker socket proxy.
- Backup `AdguardHome.yaml` at first startup, then create a second backup at each update.
- Downloads CIDR lists by country from GitHub
- Adds manual IPs from a `manually_blocked_ips.conf` file
- Updates the `AdGuardHome.yaml` file by replacing the `disallowed_clients` list
- Creates a backup of the original config (`AdGuardHome.yaml.first-start.bak`) on first run
- Creates a backup before each update (`AdGuardHome.yaml.last-update.bak`)
- Restarts the AdGuard Home container via Docker API
- Built-in Python scheduler using the `schedule` library, configurable to run updates daily or weekly
## Environment Variables
| Variable | Description | Default |
| ------------------- | ---------------------------------------------------------- | --------------------------------- |
| `TZ` | Your Time Zone | (required) |
| `BLOCK_COUNTRIES` | Comma-separated country codes to block (e.g., `CN,RU,IR`) | (required) |
| `BLOCKLIST_CRON` | Cron expression for update frequency (e.g., `0 6 * * *`) | `0 6 * * *` (at 6:00 everydays) |
| `DOCKER_API_URL` | URL of Docker socket proxy to restart AdGuard container | `http://socket-proxy-adguard:2375` |
| `ADGUARD_CONTAINER_NAME` | Name of your adguard container | `adguardhome` |
| Variable | Description | Example | Possible Values |
|--------------------------|--------------------------------------------------------------------------|-----------------------------|---------------------------------------------|
| `TZ` | Timezone of the container to correctly schedule updates | `Europe/Paris` | Any valid timezone (e.g., `UTC`, `America/New_York`, etc.) |
| `BLOCK_COUNTRIES` | List of country codes for CIDR lists, separated by commas | `cn,ru,ir` | ISO 2-letter country codes |
| `BLOCKLIST_CRON_TYPE` | Scheduling type: `daily` or `weekly` | `daily` | `daily`, `weekly` |
| `BLOCKLIST_CRON_TIME` | Time to run update in `HH:MM` 24-hour format | `06:00` | 24-hour time format |
| `BLOCKLIST_CRON_DAY` | Day of the week for weekly schedule (e.g., `mon`, `tue`, etc.) | `mon` | `mon`, `tue`, `wed`, `thu`, `fri`, `sat`, `sun` |
| `ADGUARD_CONTAINER_NAME` | Name of the AdGuard Home container to restart | `adguardhome` | Valid Docker container name |
| `DOCKER_API_URL` | Docker API URL (used to restart the container) | `http://socket-proxy-adguard:2375` | HTTP URL |
## Volumes
- `/path/to/adguard/confdir` : configuration directory containing `AdGuardHome.yaml` from your adguard container, and optionally `manually_blocked_ips.conf`.
## File Structure
@ -46,7 +55,7 @@
## Installation and Usage
### With our docker image
### With our provided docker image
1. **Create `docker-compose.yml` in your `adguard-cidre` folder**
@ -60,7 +69,10 @@
environment:
- TZ=Europe/Paris # change to your timezone
- BLOCK_COUNTRIES=cn,ru # choose countries listed IP to block. Full lists here https://github.com/vulnebify/cidre/tree/main/output/cidr/ipv4
- BLOCKLIST_CRON=0 6 * * * # at 6:00 every days
- BLOCKLIST_CRON_TYPE=daily # daily or weekly
# if weekly, choose the day
# - BLOCKLIST_CRON_DAY=mon
- BLOCKLIST_CRON_TIME=06:00
- DOCKER_API_URL=http://socket-proxy-adguard:2375 # docker socket proxy
- ADGUARD_CONTAINER_NAME=adguardhome # adguard container name
volumes:

159
blocklist_scheduler.py Normal file
View File

@ -0,0 +1,159 @@
#!/usr/bin/env python3
import os
import sys
import logging
import requests
import yaml
import schedule
import time
from pathlib import Path
logging.basicConfig(
level=logging.INFO,
format='[blocklist] %(levelname)s: %(message)s',
stream=sys.stdout,
)
ADGUARD_YAML = Path("/adguard/AdGuardHome.yaml")
TMP_YAML = ADGUARD_YAML.parent / (ADGUARD_YAML.name + ".tmp")
MANUAL_IPS_FILE = Path("/adguard/manually_blocked_ips.conf")
CIDR_BASE_URL = "https://raw.githubusercontent.com/vulnebify/cidre/main/output/cidr/ipv4"
FIRST_BACKUP = ADGUARD_YAML.parent / "AdGuardHome.yaml.first-start.bak"
LAST_UPDATE_BACKUP = ADGUARD_YAML.parent / "AdGuardHome.yaml.last-update.bak"
BLOCK_COUNTRIES = os.getenv("BLOCK_COUNTRIES", "")
BLOCKLIST_CRON_TYPE = os.getenv("BLOCKLIST_CRON_TYPE", "daily").lower() # daily or weekly
BLOCKLIST_CRON_TIME = os.getenv("BLOCKLIST_CRON_TIME", "06:00") # HH:MM format
BLOCKLIST_CRON_DAY = os.getenv("BLOCKLIST_CRON_DAY", "mon").lower() # only if weekly
ADGUARD_CONTAINER_NAME = os.getenv("ADGUARD_CONTAINER_NAME", "adguardhome")
DOCKER_API_URL = os.getenv("DOCKER_API_URL", "http://socket-proxy-adguard:2375")
def backup_first_start():
if not FIRST_BACKUP.exists():
logging.info(f"Creating first start backup: {FIRST_BACKUP}")
FIRST_BACKUP.write_text(ADGUARD_YAML.read_text())
else:
logging.info("First start backup already exists, skipping.")
def backup_last_update():
logging.info(f"Creating last update backup: {LAST_UPDATE_BACKUP}")
LAST_UPDATE_BACKUP.write_text(ADGUARD_YAML.read_text())
def download_cidr_lists(countries):
combined_ips = []
for code in countries:
url = f"{CIDR_BASE_URL}/{code.lower()}.cidr"
logging.info(f"Downloading CIDR list for {code} from {url}")
try:
r = requests.get(url, timeout=30)
r.raise_for_status()
ips = r.text.strip().splitlines()
logging.info(f"Downloaded {len(ips)} CIDR entries for {code}")
combined_ips.extend(ips)
except Exception as e:
logging.warning(f"Failed to download {code}: {e}")
return combined_ips
def read_manual_ips():
if MANUAL_IPS_FILE.exists():
logging.info(f"Reading manual IPs from {MANUAL_IPS_FILE}")
valid_ips = []
with MANUAL_IPS_FILE.open() as f:
for line in f:
line = line.strip()
if line and (line.count('.') == 3 or '/' in line):
valid_ips.append(line)
logging.info(f"Added {len(valid_ips)} manual IP entries")
return valid_ips
else:
logging.info("Manual IPs file does not exist, skipping.")
return []
def update_yaml_with_ips(ips):
if not ADGUARD_YAML.exists():
logging.error(f"{ADGUARD_YAML} does not exist. Cannot update.")
return False
data = None
with ADGUARD_YAML.open() as f:
data = yaml.safe_load(f)
if data is None:
logging.error(f"Failed to parse YAML file {ADGUARD_YAML}")
return False
data['disallowed_clients'] = ips
with TMP_YAML.open('w') as f:
yaml.safe_dump(data, f)
TMP_YAML.replace(ADGUARD_YAML)
logging.info(f"Updated {ADGUARD_YAML} with new disallowed clients list.")
return True
def restart_adguard_container():
restart_url = f"{DOCKER_API_URL}/containers/{ADGUARD_CONTAINER_NAME}/restart"
logging.info(f"Restarting AdGuard container '{ADGUARD_CONTAINER_NAME}'...")
try:
resp = requests.post(restart_url, timeout=10)
if resp.status_code == 204:
logging.info("AdGuard container restarted successfully.")
else:
logging.error(f"Failed to restart container: {resp.status_code} {resp.text}")
except Exception as e:
logging.error(f"Error restarting container: {e}")
def update_blocklist():
if not BLOCK_COUNTRIES:
logging.error("No countries specified in BLOCK_COUNTRIES environment variable. Skipping update.")
return
countries_list = [c.strip() for c in BLOCK_COUNTRIES.split(",") if c.strip()]
cidr_ips = download_cidr_lists(countries_list)
manual_ips = read_manual_ips()
combined_ips = cidr_ips + manual_ips
backup_last_update()
success = update_yaml_with_ips(combined_ips)
if success:
restart_adguard_container()
def schedule_job():
try:
hour, minute = [int(x) for x in BLOCKLIST_CRON_TIME.split(":")]
except Exception:
logging.error(f"Invalid BLOCKLIST_CRON_TIME '{BLOCKLIST_CRON_TIME}', must be HH:MM. Defaulting to 06:00.")
hour, minute = 6, 0
if BLOCKLIST_CRON_TYPE == "daily":
schedule.every().day.at(f"{hour:02d}:{minute:02d}").do(update_blocklist)
logging.info(f"Scheduled daily update at {hour:02d}:{minute:02d}")
elif BLOCKLIST_CRON_TYPE == "weekly":
valid_days = ["mon","tue","wed","thu","fri","sat","sun"]
day = BLOCKLIST_CRON_DAY[:3]
if day not in valid_days:
logging.error(f"Invalid BLOCKLIST_CRON_DAY '{BLOCKLIST_CRON_DAY}', must be one of {valid_days}. Defaulting to Monday.")
day = "mon"
getattr(schedule.every(), day).at(f"{hour:02d}:{minute:02d}").do(update_blocklist)
logging.info(f"Scheduled weekly update on {day.capitalize()} at {hour:02d}:{minute:02d}")
else:
logging.error(f"Invalid BLOCKLIST_CRON_TYPE '{BLOCKLIST_CRON_TYPE}', must be 'daily' or 'weekly'. Defaulting to daily.")
schedule.every().day.at(f"{hour:02d}:{minute:02d}").do(update_blocklist)
logging.info(f"Scheduled daily update at {hour:02d}:{minute:02d}")
def main():
logging.info("Starting blocklist scheduler...")
backup_first_start()
update_blocklist()
schedule_job()
while True:
schedule.run_pending()
time.sleep(10)
if __name__ == "__main__":
main()

View File

@ -7,7 +7,10 @@ services:
environment:
- TZ=Europe/Paris # change to your timezone
- BLOCK_COUNTRIES=cn,ru # choose countries listed IP to block. Full lists here https://github.com/vulnebify/cidre/tree/main/output/cidr/ipv4
- BLOCKLIST_CRON=0 6 * * * # at 6:00 every days
- BLOCKLIST_CRON_TYPE=daily # daily or weekly
# if weekly, choose the day
# - BLOCKLIST_CRON_DAY=mon
- BLOCKLIST_CRON_TIME=06:00
- DOCKER_API_URL=http://socket-proxy-adguard:2375 # docker socket proxy
- ADGUARD_CONTAINER_NAME=adguardhome # adguard container name
volumes:

View File

@ -1,67 +0,0 @@
#!/usr/bin/env python3
import os
import sys
import subprocess
import logging
from pathlib import Path
logging.basicConfig(
level=logging.INFO,
format='[entrypoint] %(message)s',
stream=sys.stdout
)
ADGUARD_YAML = Path("/adguard/AdGuardHome.yaml")
FIRST_BACKUP = Path("/adguard/AdGuardHome.yaml.first-start.bak")
def setup_cron():
cron_expr = os.getenv("BLOCKLIST_CRON", "0 6 * * *")
cron_line = f"{cron_expr} root /usr/local/bin/update-blocklist.py\n"
cron_dir = "/etc/crontabs"
cron_file = f"{cron_dir}/root"
logging.info(f"Setting cron job: {cron_line.strip()}")
# Ensure cron directory exists
os.makedirs(cron_dir, exist_ok=True)
with open(cron_file, "w") as f:
f.write(cron_line)
def backup_first_start():
if not FIRST_BACKUP.exists():
logging.info("Creating first start backup...")
FIRST_BACKUP.write_text(ADGUARD_YAML.read_text())
else:
logging.info("First start backup already exists.")
def run_initial_update():
logging.info("Running initial update-blocklist.py script...")
try:
subprocess.run(
["/usr/local/bin/update-blocklist.py"],
check=True,
stdout=sys.stdout,
stderr=sys.stderr,
)
except subprocess.CalledProcessError as e:
logging.error(f"Initial update script failed: {e}")
sys.exit(1)
def start_cron_foreground():
logging.info("Starting cron in foreground...")
os.execvp("cron", ["cron", "-f"])
def main():
# Check AdGuardHome.yaml exists
if not ADGUARD_YAML.exists():
logging.error(f"{ADGUARD_YAML} not found. Exiting.")
sys.exit(1)
backup_first_start()
run_initial_update()
setup_cron()
start_cron_foreground()
if __name__ == "__main__":
main()

View File

@ -1,148 +0,0 @@
#!/usr/bin/env python3
import os
import sys
import logging
import requests
from pathlib import Path
logging.basicConfig(
level=logging.INFO,
format='[update-blocklist] %(levelname)s: %(message)s',
stream=sys.stdout,
)
ADGUARD_YAML = Path("/adguard/AdGuardHome.yaml")
FIRST_BACKUP = Path("/adguard/AdGuardHome.yaml.first-start.bak")
LAST_CRON_BACKUP = Path("/adguard/AdGuardHome.yaml.last-cron.bak")
TMP_YAML = ADGUARD_YAML.parent / (ADGUARD_YAML.name + ".tmp")
MANUAL_IPS_FILE = Path("/adguard/manually_blocked_ips.conf")
CIDR_BASE_URL = "https://raw.githubusercontent.com/vulnebify/cidre/main/output/cidr/ipv4"
COUNTRIES = os.getenv("BLOCK_COUNTRIES", "")
def backup_files():
if not FIRST_BACKUP.exists():
logging.info(f"Creating first-start backup: {FIRST_BACKUP}")
FIRST_BACKUP.write_text(ADGUARD_YAML.read_text())
else:
logging.info("First-start backup already exists, skipping.")
logging.info(f"Creating last-cron backup: {LAST_CRON_BACKUP}")
LAST_CRON_BACKUP.write_text(ADGUARD_YAML.read_text())
def download_cidr_lists(countries):
combined_ips = []
for code in countries:
url = f"{CIDR_BASE_URL}/{code.lower()}.cidr"
logging.info(f"Downloading CIDR list for {code} from {url}")
try:
r = requests.get(url, timeout=30)
r.raise_for_status()
ips = r.text.strip().splitlines()
logging.info(f"Downloaded {len(ips)} CIDR entries for {code}")
combined_ips.extend(ips)
except Exception as e:
logging.warning(f"Failed to download {code}: {e}")
return combined_ips
def read_manual_ips():
if MANUAL_IPS_FILE.exists():
logging.info(f"Reading manual IPs from {MANUAL_IPS_FILE}")
valid_ips = []
with MANUAL_IPS_FILE.open() as f:
for line in f:
line = line.strip()
# Simple check for IPv4 or IPv4 CIDR format
if line and line.count('.') == 3:
valid_ips.append(line)
logging.info(f"Added {len(valid_ips)} manual IP entries")
return valid_ips
else:
logging.info("Manual IPs file does not exist, skipping.")
return []
def update_yaml_with_ips(ips):
output_lines = []
inside_disallowed = False
disallowed_indent = ""
with ADGUARD_YAML.open() as f:
lines = f.readlines()
for line in lines:
stripped = line.lstrip()
indent = line[:len(line) - len(stripped)]
if stripped.startswith("disallowed_clients:"):
# Capture the indentation of the disallowed_clients key
disallowed_indent = indent
# Replace entire line with just 'disallowed_clients:' (remove any [])
output_lines.append(f"{disallowed_indent}disallowed_clients:")
# Add all IPs indented 2 spaces more than disallowed_clients
formatted_ips = [f"{disallowed_indent} - {ip}" for ip in ips]
output_lines.extend(formatted_ips)
inside_disallowed = True
continue
if inside_disallowed:
# We skip all old lines inside disallowed_clients block.
# The block ends when we find a line with indentation
# less than or equal to disallowed_indent but not the key line itself.
# To detect end of block, compare indent length:
if len(indent) <= len(disallowed_indent) and stripped != "":
inside_disallowed = False
output_lines.append(line.rstrip("\n"))
else:
# skip this line (old disallowed_clients content)
continue
else:
output_lines.append(line.rstrip("\n"))
# Write temp file in same directory to avoid cross-device rename errors
with TMP_YAML.open("w") as f:
f.write("\n".join(output_lines) + "\n")
TMP_YAML.replace(ADGUARD_YAML)
logging.info(f"Updated {ADGUARD_YAML} with new disallowed clients list.")
def restart_adguard_container():
docker_api_url = os.getenv("DOCKER_API_URL", "http://socket-proxy-adguard:2375")
container_name = os.getenv("ADGUARD_CONTAINER_NAME", "adguardhome")
restart_url = f"{docker_api_url}/containers/{container_name}/restart"
logging.info(f"Restarting AdGuard container '{container_name}'...")
try:
resp = requests.post(restart_url, timeout=10)
if resp.status_code == 204:
logging.info("AdGuard container restarted successfully.")
else:
logging.error(f"Failed to restart container: {resp.status_code} {resp.text}")
except Exception as e:
logging.error(f"Error restarting container: {e}")
def main():
if not ADGUARD_YAML.exists():
logging.error(f"{ADGUARD_YAML} not found, exiting.")
sys.exit(1)
if not COUNTRIES:
logging.error("No countries specified in BLOCK_COUNTRIES environment variable, exiting.")
sys.exit(1)
backup_files()
countries_list = [c.strip() for c in COUNTRIES.split(",") if c.strip()]
cidr_ips = download_cidr_lists(countries_list)
manual_ips = read_manual_ips()
combined_ips = cidr_ips + manual_ips
update_yaml_with_ips(combined_ips)
restart_adguard_container()
if __name__ == "__main__":
main()