Skip to content

Commit

Permalink
Python logging support (#29)
Browse files Browse the repository at this point in the history
* Add logging support

* Support lowercase log level choices

* Update README
  • Loading branch information
WardPearce authored Sep 18, 2024
1 parent 7c10ce5 commit 8e2587e
Show file tree
Hide file tree
Showing 4 changed files with 87 additions and 45 deletions.
5 changes: 4 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,7 @@ smart-ipv6-rotator.py run [-h] [--services {google}] [--external-ipv6-ranges EXT
- `--no-services`: Completely disable the --services flag.
- `--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.
- `--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}`: Sets log level

---

Expand All @@ -85,6 +86,7 @@ smart-ipv6-rotator.py clean [-h] [--skip-root]

- `-h, --help`: Display the help message and exit.
- `--skip-root`: Skip root check.
- `--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}`: Sets log level

---

Expand All @@ -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.
- `--skip-root`: Skip root check.
- `--no-services`: Completely disable the --services flag.
- `--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}`: Sets log level

---

Expand Down Expand Up @@ -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.
### Low
- [ ] 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?
89 changes: 59 additions & 30 deletions smart_ipv6_rotator/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import argparse
import logging
import sys
from dataclasses import asdict
from ipaddress import IPv6Address, IPv6Network
Expand All @@ -14,6 +15,8 @@
IP,
IPROUTE,
LEGACY_CONFIG_FILE,
LOG_LEVELS_NAMES,
LOGGER,
)
from smart_ipv6_rotator.helpers import (
PreviousConfig,
Expand Down Expand Up @@ -58,15 +61,31 @@
"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:
params = dict(namespace.__dict__)
params.pop("subcommand")
params.pop("func")

if "log_level" in params:
LOGGER.setLevel(params["log_level"].upper())
params.pop("log_level")

return func(**params)

return _parse_args
Expand All @@ -84,12 +103,15 @@ def run(
"""Run the IPv6 rotator process."""

if path.exists(LEGACY_CONFIG_FILE):
sys.exit(
"[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"
LOGGER.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"
)
sys.exit()

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)
check_ipv6_connectivity()
Expand Down Expand Up @@ -125,9 +147,9 @@ def run(
# Save config now, will be cleaned if errors raised.
PreviousConfig(service_ranges).save(saved_ranges)

print("[DEBUG] Debug info:")
LOGGER.debug("Debug info:")
for key, value in asdict(saved_ranges).items():
print(f"{key} --> {value}")
LOGGER.debug(f"{key} --> {value}")

try:
IPROUTE.addr(
Expand All @@ -138,11 +160,12 @@ def run(
)
except Exception as error:
clean_ranges(service_ranges, skip_root)
sys.exit(
"[Error] 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"
f" Exception:\n{error}"
LOGGER.error(
"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"
f"Exception:\n{error}"
)
sys.exit()

sleep(2) # Need so that the linux kernel takes into account the new ipv6 route

Expand All @@ -159,10 +182,11 @@ def run(
)
except Exception as error:
clean_ranges(service_ranges, skip_root)
sys.exit(
"[Error] Failed to configure the test IPv6 route. The setup did not work!\n"
LOGGER.error(
"Failed to configure the test IPv6 route. The setup did not work!\n"
f" Exception:\n{error}"
)
sys.exit()

sleep(4)

Expand All @@ -174,31 +198,34 @@ def run(
)
except requests.exceptions.RequestException as error:
clean_ranges(service_ranges, skip_root)
sys.exit(
"[ERROR] 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"
" Or did you correctly configure the IPv6 subnet to use?\n"
f" Exception:\n{error}"
LOGGER.error(
"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"
"Or did you correctly configure the IPv6 subnet to use?\n"
f"Exception:\n{error}"
)
sys.exit()

try:
check_new_ipv6_address.raise_for_status()
except requests.HTTPError:
clean_ranges(service_ranges, skip_root)
sys.exit(
"[ERROR] icanhazip didn't return the expected status, possibly they are down right now."
LOGGER.error(
"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()
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:
clean_ranges(service_ranges, skip_root)
sys.exit(
"[ERROR] 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."
f" Address used: {response_new_ipv6_address}"
LOGGER.error(
"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."
f"Address used: {response_new_ipv6_address}"
)
sys.exit()

clean_ipv6_check(saved_ranges)

Expand All @@ -214,15 +241,17 @@ def run(
)
except Exception as error:
clean_ranges(service_ranges, skip_root)
sys.exit(
f"[Error] Failed to configure the service IPv6 route. The setup did not work!\n"
f" Exception:\n{error}"
LOGGER.error(
f"Failed to configure the service IPv6 route. The setup did not work!\n"
f"Exception:\n{error}"
)
sys.exit()

print(
f"[INFO] Correctly configured the IPv6 routes for IPv6 ranges {service_ranges}.\n"
"[INFO] Successful setup. Waiting for the propagation in the Linux kernel."
LOGGER.info(
f"Correctly configured the IPv6 routes for IPv6 ranges {service_ranges}.\n"
"Successful setup. Waiting for the propagation in the Linux kernel."
)

sleep(6)


Expand Down
5 changes: 5 additions & 0 deletions smart_ipv6_rotator/const.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import logging

from pyroute2 import IPDB, IPRoute

ICANHAZIP_IPV6_ADDRESS = "2606:4700::6812:7261"
Expand All @@ -6,6 +8,9 @@

LEGACY_CONFIG_FILE = "/tmp/smart-ipv6-rotator.py"

LOGGER = logging.getLogger(__name__)
LOG_LEVELS_NAMES = list(logging._nameToLevel.keys())

IP = IPDB()
IPROUTE = IPRoute()

Expand Down
33 changes: 19 additions & 14 deletions smart_ipv6_rotator/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,18 @@
import os
import sys
from dataclasses import asdict
from requests.adapters import HTTPAdapter
from time import sleep
from typing import Iterator

import requests
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.ranges import RANGES

Expand All @@ -22,16 +26,18 @@ def root_check(skip_root: bool = False) -> None:
def check_ipv6_connectivity() -> None:
try:
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)
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:
sys.exit(
"[ERROR] icanhazip didn't return the expected status, possibly they are down right now."
LOGGER.error(
"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(
Expand Down Expand Up @@ -95,7 +101,7 @@ def clean_ranges(ranges_: list[str], skip_root: bool) -> None:

previous = previous_config.get()
if not previous:
print("[INFO] No cleanup of previous setup needed.")
LOGGER.info("No cleanup of previous setup needed.")
return

clean_ipv6_check(previous)
Expand All @@ -110,12 +116,11 @@ def clean_ranges(ranges_: list[str], skip_root: bool) -> None:
oif=previous.interface_index,
)
except:
print(
f"""[Error] Failed to remove the configured IPv6 subnets {','.join(previous.ranges)}
LOGGER.error(
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
"""
)

try:
IPROUTE.addr(
"del",
Expand All @@ -124,12 +129,12 @@ def clean_ranges(ranges_: list[str], skip_root: bool) -> None:
mask=previous.random_ipv6_address_mask,
)
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()

print(
"[INFO] Finished cleaning up previous setup.\n[INFO] Waiting for the propagation in the Linux kernel."
LOGGER.info(
"Finished cleaning up previous setup.\nWaiting for the propagation in the Linux kernel."
)

sleep(6)
Expand Down

0 comments on commit 8e2587e

Please sign in to comment.