Proxy failover with Freebox API
Problem Explanation
Behind an IP address (and so behind my Freebox), I can only forward traffic on ports 80 and 443 to a single IP at any given time. This limitation means that only one server can act as the proxy for all the local services at a time. Before the more powerful "Server 1" was serving as the proxy. However, when Server 1 goes down (it happens sometimes, this is not production here !!) all the services on the "Server 2" are also down, which is not cool. So I needed this server to takes over by updating the router's port forwarding settings and launching the Nginx Proxy Manager Docker container. Server 1 ensures that Server 2 is always ready to take over by transferring the container's files daily at 2 AM. This failover mechanism ensures continuous availability of the proxy service despite the single IP forwarding constraint.
In this guide, we will walk through how to set up a failover mechanism with a python script and a cronjob for Nginx Proxy Manager (NPM) using the Freebox API. The failover setup ensures that if the primary NPM instance goes down, traffic is automatically redirected to a backup instance, maintaining service availability.
Nginx Proxy Manager Installation
On each machine we launch at least one time the NPM container, using a simple Docker Compose. Create a docker-compose.yml file in a directory named nginx-proxy-manager.
services:
nginx-proxy-manager:
image: 'jc21/nginx-proxy-manager:latest'
restart: always
security_opt:
- label:disable
ports:
- '80:80' # Public HTTP Port
- '443:443' # Public HTTPS Port
- '8081:81' # Admin Web Port
volumes:
- ./container-data:/data
- ./certs:/etc/letsencrypt
To start the NPM container, run:
docker-compose up -d
Once the container is up, access the NPM interface at <your-IP>:8081 to configure your proxy hosts. All the configuration of NPM and Domain name will be part of another post.
Setting Up Failover with Freebox API
Get your api token
To set up the failover system, you need to authorize your application with the Freebox API and obtain the necessary tokens.
- Request Authorization: Modify the python script to match your app name. Then launch the script, this will generate an application token and a track ID. The Freebox will prompt you to confirm the authorization on the device. You only need to click OK (depend on the Freebox version)
python3 get_token.py
import requests
import time
import hmac
import hashlib
import json
# Define the application details
APP_ID = "com.yourapp.com"
APP_NAME = "FailoverNPM"
APP_VERSION = "0.0.1"
DEVICE_NAME = "backupserver"
FREEBOX_API_URL = "http://mafreebox.freebox.fr/api/v4"
def request_authorization():
url = f"{FREEBOX_API_URL}/login/authorize/"
payload = {
"app_id": APP_ID,
"app_name": APP_NAME,
"app_version": APP_VERSION,
"device_name": DEVICE_NAME
}
response = requests.post(url, json=payload)
response_data = response.json()
if response_data['success']:
return response_data['result']['app_token'], response_data['result']['track_id']
else:
raise Exception("Authorization request failed")
def track_authorization(track_id):
url = f"{FREEBOX_API_URL}/login/authorize/{track_id}"
while True:
response = requests.get(url)
response_data = response.json()
if response_data['success']:
status = response_data['result']['status']
if status == 'pending':
print("Authorization pending, please confirm on the Freebox...")
time.sleep(30) # Wait for 5 seconds before retrying
elif status == 'granted':
return response_data['result']['challenge']
else:
raise Exception(f"Authorization failed with status: {status}")
else:
raise Exception("Failed to track authorization")
def get_session_token(app_token, challenge):
url = f"{FREEBOX_API_URL}/login/session/"
password = hmac.new(app_token.encode(), challenge.encode(), hashlib.sha1).hexdigest()
payload = {
"app_id": APP_ID,
"app_version": APP_VERSION,
"password": password
}
response = requests.post(url, json=payload)
response_data = response.json()
if response_data['success']:
return response_data['result']['session_token']
else:
raise Exception("Failed to obtain session token")
def main():
try:
# Step 1: Request authorization
app_token, track_id = request_authorization()
print(f"App Token: {app_token}, Track ID: {track_id}")
# Step 2: Track authorization progress
challenge = track_authorization(track_id)
print(f"Authorization granted, challenge: {challenge}")
# Step 3: Obtain session token
session_token = get_session_token(app_token, challenge)
print(f"Session Token: {session_token}")
# Save the app token to token.txt
with open("token.txt", "w") as token_file:
token_file.write(app_token)
print("App token saved to token.txt")
except Exception as e:
print(f"An error occurred: {e}")
if __name__ == "__main__":
main()
Failover Script
- Put the tokens on freebox_tokens.txt file
yourtoken
your_session_token #optional will be put after to reuse the session will it is open
- Edit the Crontab: Open the crontab editor to set up your cron job.
crontab -e
- Add the Cron Job: Add the following line to the crontab file to schedule the script to run every 5 minutes.
*/5 * * * * /usr/bin/python3 /docker/failover/failover.py
This line specifies that the run_failover.sh script should run every 5 minutes. The output (including any errors) will be logged to /docker/failover/logs/failover_script.log for monitoring purposes.
Breakdown of the code
- Configuration: Define IP addresses, ports, and Freebox API credentials.
- Logging: Set up logging to track the script's actions.
- Token Management: Functions to read, write, and obtain Freebox session tokens.
- NPM Status Check: Check if NPM is running on a given IP and port.
- Redirection Management: Add, delete, and verify port redirections on the Freebox router.
- Docker Control: Start or stop Docker containers based on the NPM status.
import requests
import json
import time
import hmac
import hashlib
import subprocess
import argparse
import os
import logging
from logging.handlers import RotatingFileHandler
# Configuration
MASTER_IP = "SERVEUR_1_IP"
FAILOVER_IP = "SERVEUR_2_IP"
NPM_PORT = 81
PROXY_PORTS = [80, 443]
PROTOCOLS = ["tcp", "udp"]
API_BASE_URL = "https://[your_freebox_private_base].fbxos.fr:8205/api/v4"
APP_ID = "[YOUR_APP_ID]"
APP_NAME = "[YOUR_APP_NAME]"
APP_VERSION = "0.0.1"
DEVICE_NAME = "backupserver"
DOCKER_IMAGE = "jc21/nginx-proxy-manager:latest"
DOCKER_CONTAINER_NAME = "nginx-proxy-manager"
CERTIFICATE_FILE = "/docker/failover/rootca.pem"
# Parse command-line arguments
parser = argparse.ArgumentParser()
parser.add_argument('-d', '--debug', action='store_true', help='Enable debug mode')
parser.add_argument('-f', '--file', default='/docker/failover/freebox_tokens.txt', help='Path to freebox_tokens.txt')
args = parser.parse_args()
# Set up logging
log_file = '/docker/failover/logs/failover_script.log'
max_log_size = 10 * 1024 * 1024 # 10MB
backup_count = 5
handler = RotatingFileHandler(log_file, maxBytes=max_log_size, backupCount=backup_count)
logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[handler])
def get_session_token():
app_token, session_token = read_tokens()
if session_token:
# Check if the session token is still valid
headers = {"X-Fbx-App-Auth": session_token}
response = requests.get(f"{API_BASE_URL}/connection/", headers=headers, verify=CERTIFICATE_FILE)
if response.status_code == 200:
return session_token
# If we don't have a valid session token, get a new one
challenge = get_challenge()
password = hmac.new(app_token.encode(), challenge.encode(), hashlib.sha1).hexdigest()
payload = {
"app_id": APP_ID,
"app_version": APP_VERSION,
"password": password
}
response = requests.post(f"{API_BASE_URL}/login/session/", json=payload, verify=CERTIFICATE_FILE)
response_data = response.json()
if response_data['success']:
new_session_token = response_data['result']['session_token']
write_tokens(app_token, new_session_token)
return new_session_token
else:
raise Exception("Failed to obtain session token")
def read_tokens():
try:
with open(args.file, 'r') as file:
app_token = file.readline().strip()
session_token = file.readline().strip()
return app_token, session_token
except FileNotFoundError:
return None, None
def write_tokens(app_token, session_token):
with open(args.file, 'w') as file:
file.write(f"{app_token}\n{session_token}")
def get_challenge():
url = f"{API_BASE_URL}/login/"
response = requests.get(url, verify=CERTIFICATE_FILE)
response_data = response.json()
if response_data['success']:
return response_data['result']['challenge']
else:
raise Exception("Failed to obtain challenge")
def check_nginx_proxy_manager(ip, port):
url = f"http://{ip}:{port}"
try:
response = requests.get(url, timeout=5)
return response.status_code == 200
except requests.RequestException:
return False
def get_redirections(session_token):
headers = {"X-Fbx-App-Auth": session_token}
response = requests.get(f"{API_BASE_URL}/fw/redir/", headers=headers, verify=CERTIFICATE_FILE)
return response.json()
def delete_redirection(redir_id, session_token):
headers = {"X-Fbx-App-Auth": session_token}
response = requests.delete(f"{API_BASE_URL}/fw/redir/{redir_id}", headers=headers, verify=CERTIFICATE_FILE)
return response.json()
def add_redirection(port, protocol, target_ip, session_token):
headers = {
"X-Fbx-App-Auth": session_token,
"Content-Type": "application/json"
}
data = {
"enabled": True,
"lan_port": port,
"wan_port_start": port,
"wan_port_end": port,
"lan_ip": target_ip,
"ip_proto": protocol,
"src_ip": "0.0.0.0",
"comment": f"Redirection for port {port} {protocol}"
}
response = requests.post(f"{API_BASE_URL}/fw/redir/", headers=headers, json=data, verify=CERTIFICATE_FILE)
return response.json()
def manage_redirections(target_ip):
ports_modified = False
session_token = get_session_token()
redirections = get_redirections(session_token)
for port in PROXY_PORTS:
for protocol in PROTOCOLS:
logging.debug(f"Processing port {port} {protocol}")
redir_id = None
current_ip = None
for redirection in redirections.get('result', []):
if redirection['wan_port_start'] == port and redirection['ip_proto'] == protocol:
redir_id = redirection['id']
current_ip = redirection['lan_ip']
break
if current_ip == target_ip:
logging.debug(f"Redirection for port {port} {protocol} to {target_ip} already exists. Skipping.")
elif redir_id:
ports_modified = True
logging.debug(f"Deleting redirection {redir_id} for port {port} {protocol} (current IP: {current_ip})")
delete_redirection(redir_id, session_token)
logging.debug(f"Adding redirection for port {port} {protocol} to {target_ip}")
add_redirection(port, protocol, target_ip, session_token)
else:
ports_modified = True
logging.debug(f"Adding new redirection for port {port} {protocol} to {target_ip}")
add_redirection(port, protocol, target_ip, session_token)
if ports_modified:
logging.info(f"Port forwarding applied for {target_ip}")
else:
logging.info(f"Ports are already well configured")
def docker_container(action):
command = f"docker compose -f /docker/docker-compose.yml {action} proxy"
subprocess.run(command, shell=True)
def main():
if check_nginx_proxy_manager(MASTER_IP, NPM_PORT):
logging.info(f"N
