Python logging support (#29)
Some checks failed
Docker Multi-Architecture Build / build-docker (push) Failing after 1s

* Add logging support

* Support lowercase log level choices

* Update README
This commit is contained in:
Ward 2024-09-19 09:00:53 +12:00 committed by GitHub
parent 7c10ce5eb8
commit 8e2587e1da
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 87 additions and 45 deletions

View File

@ -72,6 +72,7 @@ smart-ipv6-rotator.py run [-h] [--services {google}] [--external-ipv6-ranges EXT
- `--no-services`: Completely disable the --services flag. - `--no-services`: Completely disable the --services flag.
- `--ipv6range IPV6RANGE`: Your IPV6 range (e.g., 2407:7000:9827:4100::/64). - `--ipv6range IPV6RANGE`: Your IPV6 range (e.g., 2407:7000:9827:4100::/64).
- `--cron`: Do not check if the IPv6 address configured will work properly. Useful for CRON and when you know that the IPv6 range is correct. - `--cron`: Do not check if the IPv6 address configured will work properly. Useful for CRON and when you know that the IPv6 range is correct.
- `--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}`: Sets log level
--- ---
@ -85,6 +86,7 @@ smart-ipv6-rotator.py clean [-h] [--skip-root]
- `-h, --help`: Display the help message and exit. - `-h, --help`: Display the help message and exit.
- `--skip-root`: Skip root check. - `--skip-root`: Skip root check.
- `--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}`: Sets log level
--- ---
@ -101,6 +103,7 @@ smart-ipv6-rotator.py clean-one [-h] [--services {google}] [--external-ipv6-rang
- `--external-ipv6-ranges EXTERNAL_IPV6_RANGES`: Manually define external IPV6 ranges to rotate for. - `--external-ipv6-ranges EXTERNAL_IPV6_RANGES`: Manually define external IPV6 ranges to rotate for.
- `--skip-root`: Skip root check. - `--skip-root`: Skip root check.
- `--no-services`: Completely disable the --services flag. - `--no-services`: Completely disable the --services flag.
- `--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}`: Sets log level
--- ---
@ -128,5 +131,5 @@ The attack surface of this script is very limited as it is not running in the ba
- [ ] In most time, adding the new random IPv6 will take precedence over the existing IPv6. This may not be the expected behavior. - [ ] In most time, adding the new random IPv6 will take precedence over the existing IPv6. This may not be the expected behavior.
### Low ### Low
- [ ] Argument for testing if the setup will work without permanently do any modification. - [ ] Argument for testing if the setup will work without permanently do any modification.
- [ ] Allow to remove debug info - [X] Allow to remove debug info
- [ ] Maybe not depend on icanhazip? Send requests in HTTPS? - [ ] Maybe not depend on icanhazip? Send requests in HTTPS?

View File

@ -1,4 +1,5 @@
import argparse import argparse
import logging
import sys import sys
from dataclasses import asdict from dataclasses import asdict
from ipaddress import IPv6Address, IPv6Network from ipaddress import IPv6Address, IPv6Network
@ -14,6 +15,8 @@ from smart_ipv6_rotator.const import (
IP, IP,
IPROUTE, IPROUTE,
LEGACY_CONFIG_FILE, LEGACY_CONFIG_FILE,
LOG_LEVELS_NAMES,
LOGGER,
) )
from smart_ipv6_rotator.helpers import ( from smart_ipv6_rotator.helpers import (
PreviousConfig, PreviousConfig,
@ -58,15 +61,31 @@ SHARED_OPTIONS = [
"help": "Completely disables the --services flag.", "help": "Completely disables the --services flag.",
}, },
), ),
(
"--log-level",
{
"type": str,
"choices": LOG_LEVELS_NAMES
+ [log_level.lower() for log_level in LOG_LEVELS_NAMES],
"default": "DEBUG",
"help": f"Sets log level, can be {','.join(LOG_LEVELS_NAMES)}",
},
),
] ]
logging.basicConfig(format="%(levelname)s:%(name)s:%(message)s")
def parse_args(func) -> Callable[..., Any]:
def parse_args(func: Callable) -> Callable[..., Any]:
def _parse_args(namespace: argparse.Namespace) -> Any: def _parse_args(namespace: argparse.Namespace) -> Any:
params = dict(namespace.__dict__) params = dict(namespace.__dict__)
params.pop("subcommand") params.pop("subcommand")
params.pop("func") params.pop("func")
if "log_level" in params:
LOGGER.setLevel(params["log_level"].upper())
params.pop("log_level")
return func(**params) return func(**params)
return _parse_args return _parse_args
@ -84,12 +103,15 @@ def run(
"""Run the IPv6 rotator process.""" """Run the IPv6 rotator process."""
if path.exists(LEGACY_CONFIG_FILE): if path.exists(LEGACY_CONFIG_FILE):
sys.exit( LOGGER.error(
"[ERROR] Legacy database format detected! Please run `python smart-ipv6-rotator.py clean` using the old version of this script.\nhttps://github.com/iv-org/smart-ipv6-rotator" "Legacy database format detected! Please run `python smart-ipv6-rotator.py clean` using the old version of this script.\nhttps://github.com/iv-org/smart-ipv6-rotator"
) )
sys.exit()
if cron is True: if cron is True:
print("[INFO] Running without checking if the IPv6 address configured will work properly.") LOGGER.info(
"Running without checking if the IPv6 address configured will work properly."
)
root_check(skip_root) root_check(skip_root)
check_ipv6_connectivity() check_ipv6_connectivity()
@ -125,9 +147,9 @@ def run(
# Save config now, will be cleaned if errors raised. # Save config now, will be cleaned if errors raised.
PreviousConfig(service_ranges).save(saved_ranges) PreviousConfig(service_ranges).save(saved_ranges)
print("[DEBUG] Debug info:") LOGGER.debug("Debug info:")
for key, value in asdict(saved_ranges).items(): for key, value in asdict(saved_ranges).items():
print(f"{key} --> {value}") LOGGER.debug(f"{key} --> {value}")
try: try:
IPROUTE.addr( IPROUTE.addr(
@ -138,11 +160,12 @@ def run(
) )
except Exception as error: except Exception as error:
clean_ranges(service_ranges, skip_root) clean_ranges(service_ranges, skip_root)
sys.exit( LOGGER.error(
"[Error] Failed to add the new random IPv6 address. The setup did not work!\n" "Failed to add the new random IPv6 address. The setup did not work!\n"
" That's unexpected! Did you correctly configure the IPv6 subnet to use?\n" "That's unexpected! Did you correctly configure the IPv6 subnet to use?\n"
f" Exception:\n{error}" f"Exception:\n{error}"
) )
sys.exit()
sleep(2) # Need so that the linux kernel takes into account the new ipv6 route sleep(2) # Need so that the linux kernel takes into account the new ipv6 route
@ -159,10 +182,11 @@ def run(
) )
except Exception as error: except Exception as error:
clean_ranges(service_ranges, skip_root) clean_ranges(service_ranges, skip_root)
sys.exit( LOGGER.error(
"[Error] Failed to configure the test IPv6 route. The setup did not work!\n" "Failed to configure the test IPv6 route. The setup did not work!\n"
f" Exception:\n{error}" f" Exception:\n{error}"
) )
sys.exit()
sleep(4) sleep(4)
@ -174,31 +198,34 @@ def run(
) )
except requests.exceptions.RequestException as error: except requests.exceptions.RequestException as error:
clean_ranges(service_ranges, skip_root) clean_ranges(service_ranges, skip_root)
sys.exit( LOGGER.error(
"[ERROR] Failed to send the request for checking the new IPv6 address! The setup did not work!\n" "Failed to send the request for checking the new IPv6 address! The setup did not work!\n"
" Your provider probably does not allow setting any arbitrary IPv6 address.\n" "Your provider probably does not allow setting any arbitrary IPv6 address.\n"
" Or did you correctly configure the IPv6 subnet to use?\n" "Or did you correctly configure the IPv6 subnet to use?\n"
f" Exception:\n{error}" f"Exception:\n{error}"
) )
sys.exit()
try: try:
check_new_ipv6_address.raise_for_status() check_new_ipv6_address.raise_for_status()
except requests.HTTPError: except requests.HTTPError:
clean_ranges(service_ranges, skip_root) clean_ranges(service_ranges, skip_root)
sys.exit( LOGGER.error(
"[ERROR] icanhazip didn't return the expected status, possibly they are down right now." "icanhazip didn't return the expected status, possibly they are down right now."
) )
sys.exit()
response_new_ipv6_address = check_new_ipv6_address.text.strip() response_new_ipv6_address = check_new_ipv6_address.text.strip()
if response_new_ipv6_address == random_ipv6_address: if response_new_ipv6_address == random_ipv6_address:
print("[INFO] Correctly using the new random IPv6 address, continuing.") LOGGER.info("Correctly using the new random IPv6 address, continuing.")
else: else:
clean_ranges(service_ranges, skip_root) clean_ranges(service_ranges, skip_root)
sys.exit( LOGGER.error(
"[ERROR] The new random IPv6 is not used! The setup did not work!\n" "The new random IPv6 is not used! The setup did not work!\n"
" That is very unexpected, check if your IPv6 routes do not have too much priority." "That is very unexpected, check if your IPv6 routes do not have too much priority."
f" Address used: {response_new_ipv6_address}" f"Address used: {response_new_ipv6_address}"
) )
sys.exit()
clean_ipv6_check(saved_ranges) clean_ipv6_check(saved_ranges)
@ -214,15 +241,17 @@ def run(
) )
except Exception as error: except Exception as error:
clean_ranges(service_ranges, skip_root) clean_ranges(service_ranges, skip_root)
sys.exit( LOGGER.error(
f"[Error] Failed to configure the service IPv6 route. The setup did not work!\n" f"Failed to configure the service IPv6 route. The setup did not work!\n"
f" Exception:\n{error}" f"Exception:\n{error}"
) )
sys.exit()
print( LOGGER.info(
f"[INFO] Correctly configured the IPv6 routes for IPv6 ranges {service_ranges}.\n" f"Correctly configured the IPv6 routes for IPv6 ranges {service_ranges}.\n"
"[INFO] Successful setup. Waiting for the propagation in the Linux kernel." "Successful setup. Waiting for the propagation in the Linux kernel."
) )
sleep(6) sleep(6)

View File

@ -1,3 +1,5 @@
import logging
from pyroute2 import IPDB, IPRoute from pyroute2 import IPDB, IPRoute
ICANHAZIP_IPV6_ADDRESS = "2606:4700::6812:7261" ICANHAZIP_IPV6_ADDRESS = "2606:4700::6812:7261"
@ -6,6 +8,9 @@ JSON_CONFIG_FILE = "/tmp/smart-ipv6-rotator.json"
LEGACY_CONFIG_FILE = "/tmp/smart-ipv6-rotator.py" LEGACY_CONFIG_FILE = "/tmp/smart-ipv6-rotator.py"
LOGGER = logging.getLogger(__name__)
LOG_LEVELS_NAMES = list(logging._nameToLevel.keys())
IP = IPDB() IP = IPDB()
IPROUTE = IPRoute() IPROUTE = IPRoute()

View File

@ -2,14 +2,18 @@ import json
import os import os
import sys import sys
from dataclasses import asdict from dataclasses import asdict
from requests.adapters import HTTPAdapter
from time import sleep from time import sleep
from typing import Iterator from typing import Iterator
import requests import requests
from requests.adapters import HTTPAdapter from requests.adapters import HTTPAdapter
from smart_ipv6_rotator.const import ICANHAZIP_IPV6_ADDRESS, IPROUTE, JSON_CONFIG_FILE from smart_ipv6_rotator.const import (
ICANHAZIP_IPV6_ADDRESS,
IPROUTE,
JSON_CONFIG_FILE,
LOGGER,
)
from smart_ipv6_rotator.models import SavedRanges from smart_ipv6_rotator.models import SavedRanges
from smart_ipv6_rotator.ranges import RANGES from smart_ipv6_rotator.ranges import RANGES
@ -22,16 +26,18 @@ def root_check(skip_root: bool = False) -> None:
def check_ipv6_connectivity() -> None: def check_ipv6_connectivity() -> None:
try: try:
s = requests.Session() s = requests.Session()
s.mount('http://', HTTPAdapter(max_retries=3)) s.mount("http://", HTTPAdapter(max_retries=3))
s.get("http://ipv6.icanhazip.com", timeout=10) s.get("http://ipv6.icanhazip.com", timeout=10)
except requests.Timeout: except requests.Timeout:
sys.exit("[Error] You do not have IPv6 connectivity. This script can not work.") LOGGER.error("You do not have IPv6 connectivity. This script can not work.")
sys.exit()
except requests.HTTPError: except requests.HTTPError:
sys.exit( LOGGER.error(
"[ERROR] icanhazip didn't return the expected status, possibly they are down right now." "icanhazip didn't return the expected status, possibly they are down right now."
) )
sys.exit()
print("[INFO] You have IPv6 connectivity. Continuing.") LOGGER.info("You have IPv6 connectivity. Continuing.")
def what_ranges( def what_ranges(
@ -95,7 +101,7 @@ def clean_ranges(ranges_: list[str], skip_root: bool) -> None:
previous = previous_config.get() previous = previous_config.get()
if not previous: if not previous:
print("[INFO] No cleanup of previous setup needed.") LOGGER.info("No cleanup of previous setup needed.")
return return
clean_ipv6_check(previous) clean_ipv6_check(previous)
@ -110,12 +116,11 @@ def clean_ranges(ranges_: list[str], skip_root: bool) -> None:
oif=previous.interface_index, oif=previous.interface_index,
) )
except: except:
print( LOGGER.error(
f"""[Error] Failed to remove the configured IPv6 subnets {','.join(previous.ranges)} f"""Failed to remove the configured IPv6 subnets {','.join(previous.ranges)}
May be expected if the route were not yet configured and that was a cleanup due to an error May be expected if the route were not yet configured and that was a cleanup due to an error
""" """
) )
try: try:
IPROUTE.addr( IPROUTE.addr(
"del", "del",
@ -124,12 +129,12 @@ def clean_ranges(ranges_: list[str], skip_root: bool) -> None:
mask=previous.random_ipv6_address_mask, mask=previous.random_ipv6_address_mask,
) )
except: except:
print("[Error] Failed to remove the random IPv6 address, very unexpected!") LOGGER.error("Failed to remove the random IPv6 address, very unexpected!")
previous_config.remove() previous_config.remove()
print( LOGGER.info(
"[INFO] Finished cleaning up previous setup.\n[INFO] Waiting for the propagation in the Linux kernel." "Finished cleaning up previous setup.\nWaiting for the propagation in the Linux kernel."
) )
sleep(6) sleep(6)