5
\$\begingroup\$

I needed a way to automatically update my IP forwarding rules on Google Domains for my home web server running behind my router. (I'm not paying for a static IP!). After writing a simple functional script for myself, I decided to refactor it to be object oriented and make it work for anyone.

I know you could write a simpler bash script to do something similar, but a) I love Python and wouldn't get past the shebang in bash without a lot of googling, lol, and b) I actually don't think it would be as effective.

It's pretty simple I guess, but I would welcome any feedback/criticism as I'm relatively new to all this.

ipchecker.py
import base64
import getopt
import logging
import os
import pickle
import smtplib
import sys
from email.errors import MessageError
from email.message import EmailMessage
from getpass import getpass
from itertools import cycle

from requests import get, post
from requests.exceptions import ConnectionError as ReqConError


def get_cwd():

    """Change &/ return working directory depending on OS: for absolute file paths
    Cron jobs will have '/' as their working dir by default."""

    if os.name == 'nt':
        return os.path.dirname(os.path.realpath(__file__))
    os.chdir(os.path.dirname(os.path.abspath(__file__)))
    return os.getcwd()


FILE_PATH = get_cwd()
USER_PICKLE = '%s/.user' % FILE_PATH
LOG_FILE = '%s/ipchecker.log' % FILE_PATH

logger = logging.getLogger('')
logger.setLevel(logging.INFO)
fh = logging.FileHandler(LOG_FILE)
sh = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter('[%(levelname)s]|%(asctime)s|%(message)s',
                              datefmt='%d %b %Y %H:%M:%S')
fh.setFormatter(formatter)
sh.setFormatter(formatter)
logger.addHandler(sh)
logger.addHandler(fh)


class User:
    BASE_URL = '@domains.google.com/nic/update?hostname='

    def __init__(self):

        """Create user instance and save it for future changes to API and for email notifications,
        or load previous user profile"""

        if os.path.isfile(USER_PICKLE):
            self.__dict__.update(self.load_user().__dict__)
        else:
            self.notifications = 'Y'
            self.domain, self.DNS_username, self.DNS_password, self.req_url = self.set_credentials()
            self.gmail_address, self.gmail_password = self.set_email()
            self.save_user()
            logging.info('New user created. (See `python3 ipchecker.py --help` for help changing/removing the user)')

    def set_credentials(self, update=False):

        """Set/return attributes for Google Domains credentials"""

        self.domain = input("What's your domain? (example.com / subdomain.example.com): ")
        self.DNS_username = input("What's your autogenerated DNS username?: ")
        self.DNS_password = input("What's your autogenerated DNS password?: ")
        self.req_url = f'https://{self.DNS_username}:{self.DNS_password}{self.BASE_URL}{self.domain}&myip='
        if update:
            self.save_user()
        return self.domain, self.DNS_username, self.DNS_password, self.req_url

    def set_email(self):

        """Set/return attributes for Gmail credentials if user enables notifications"""

        self.notifications = input("Enable email notifications? [Y]all(default); [e]errors only; [n]no: ").lower()
        if self.notifications != 'n':
            self.gmail_address = input("What's your email address?: ")
            self.gmail_password = base64.b64encode(getpass("What's your email password?: ").encode("utf-8"))
            return self.gmail_address, self.gmail_password
        else:
            return None, None

    def send_notification(self, ip, msg_type='success', error=None):

        """Notify user via email if IP change is made successfully or if API call fails."""

        if self.notifications != 'n':
            msg = EmailMessage()
            if msg_type == 'success' and self.notifications not in ('n', 'e'):
                msg.set_content(f'IP for {self.domain} has changed! New IP: {ip}')
                msg['Subject'] = 'IP CHANGED SUCCESSFULLY!'
            elif msg_type == 'error' and self.notifications != 'n':
                msg.set_content(f'IP for {self.domain} has changed but the API call failed ({error})! New IP: {ip}')
                msg['Subject'] = 'IP CHANGE FAILED!'
            msg['From'] = self.gmail_address
            msg['To'] = self.gmail_address
            try:
                server = smtplib.SMTP_SSL('smtp.gmail.com', 465)
                server.ehlo()
                server.login(self.gmail_address, base64.b64decode(self.gmail_password).decode('utf-8'))
                server.send_message(msg)
                server.close()
            except (MessageError, ConnectionError) as e:
                log_msg = 'Email notification not sent: %s' % e
                logger.warning(log_msg)

    def save_user(self):
        with open(USER_PICKLE, 'wb') as pickle_file:
            pickle.dump(self, pickle_file)

    @staticmethod
    def load_user(pickle_file=USER_PICKLE):
        with open(pickle_file, 'rb') as user_pickle:
            return pickle.load(user_pickle)

    @staticmethod
    def delete_user():
        os.remove(USER_PICKLE)


class IPChanger:

    def __init__(self, argv):

        """Check for command line arguments, load User instance,
        check previous IP address against current external IP, and change if different."""

        opts = []
        try:
            self.current_ip = get('https://api.ipify.org').text
            opts, _args = getopt.getopt(argv,
                                        'cdehnu:',
                                        ['credentials',
                                         'delete_user',
                                         'email',
                                         'help',
                                         'notifications',
                                         'user_load='])
        except getopt.GetoptError:
            print('ipchecker.py -h --help')
            sys.exit(2)
        except ReqConError:
            logger.warning('Connection Error')
            sys.exit(5)
        finally:
            if opts:
                for opt, arg in opts:
                    if opt in ('-h', '--help'):
                        print(
                            """
        ipChecker.py help manual (command line options):
        ``````````````````````````````````````````````````````````````````````````````````````````````````````````
        ipchecker.py                        || -run the script normally without arguments
        ipchecker.py -h --help              || -show this help manual
        ipchecker.py -c --credentials       || -change API credentials
        ipchecker.py -e --email             || -email set up wizard > use to delete email credentials (choose 'n')
        ipchecker.py -n --notifications     || -toggle email notification settings > will not delete email address
        ipchecker.py -d --delete_user       || -delete current user profile
        ipchecker.py -u path/to/user.file   || (or `--user_load user.file`) -load user from file**
                                            || **this will overwrite any current user profile without warning!
                                            || **Backup "/.user" file to store multiple profiles.
"""
                        )
                    elif opt in ('-c', '--credentials'):
                        self.user = User()
                        self.user.set_credentials(update=True)
                        self.domains_api_call()
                        logger.info('***API credentials changed***')
                    elif opt in ('-d', '--delete'):
                        User.delete_user()
                        logger.info('***User deleted***')
                        print('>>>Run the script without options to create a new user, or '
                              '`ipchecker.py -u path/to/pickle` to load one from file')
                    elif opt in ('-e', '--email'):
                        self.user = User()
                        self.user.set_email()
                        self.user.save_user()
                        logger.info('***Notification settings changed***')
                    elif opt in ('-n', '--notifications'):
                        n_options = {'Y': '[all changes]', 'e': '[errors only]', 'n': '[none]'}
                        self.user = User()
                        options_iter = cycle(n_options.keys())
                        for option in options_iter:
                            if self.user.notifications == option:
                                break
                        self.user.notifications = next(options_iter)
                        self.user.save_user()
                        log_msg = '***Notification settings changed to %s***' % n_options[self.user.notifications]
                        logger.info(log_msg)
                        if self.user.notifications in ('Y', 'e') and not self.user.gmail_address:
                            logger.info('No email user set, running email set up wizard from beginning...')
                            self.user.set_email()
                            self.user.save_user()
                    elif opt in ('-u', '--user_load'):
                        try:
                            self.user = User.load_user(pickle_file=arg)
                            self.user.save_user()
                            logger.info('***User loaded***')
                        except FileNotFoundError as e:
                            logger.warning(e)
                            sys.exit(2)
                    sys.exit()
            try:
                self.user = User()
                if self.user.previous_ip == self.current_ip:
                    log_msg = 'Current IP: %s (no change)' % self.user.previous_ip
                    logger.info(log_msg)
                else:
                    self.user.previous_ip = self.current_ip
                    self.domains_api_call()
                    log_msg = 'Newly recorded IP: %s' % self.user.previous_ip
                    logger.info(log_msg)
                    self.user.save_user()
            except AttributeError:
                setattr(self.user, 'previous_ip', self.current_ip)
                self.user.save_user()
                self.domains_api_call()
            finally:
                sys.exit()

    def domains_api_call(self):

        """Attempt to change the Dynamic DNS rules via the Google Domains API and handle response codes"""

        try:
            req = post(f'{self.user.req_url}{self.current_ip}')
            response = req.content.decode('utf-8')
            log_msg = 'Google Domains API response: %s' % response
            logger.info(log_msg)

            # Successful request:
            if response[:4] == 'good' or response[:5] == 'nochg':
                self.user.send_notification(self.current_ip)

            # Unsuccessful requests:
            elif response == 'nohost' or response == 'notfqdn':
                msg = "The hostname does not exist, is not a fully qualified domain" \
                      " or does not have Dynamic DNS enabled. The script will not be " \
                      "able to run until you fix this. See https://support.google.com/domains/answer/6147083?hl=en-CA" \
                      " for API documentation"
                logger.warning(msg)
                if input("Recreate the API profile? (Y/n):").lower() != 'n':
                    self.user.set_credentials(update=True)
                    self.domains_api_call()
                else:
                    self.user.send_notification(self.current_ip, 'error', msg)
            else:
                logger.warning("Could not authenticate with these credentials")
                if input("Recreate the API profile? (Y/n):").lower() != 'n':
                    self.user.set_credentials(update=True)
                    self.domains_api_call()
                else:
                    self.user.delete_user()
                    logger.warning('API authentication failed, user profile deleted')
                    sys.exit(1)
        # Local connection related errors:
        except ReqConError as e:
            log_msg = 'Connection Error: %s' % e
            logger.warning(log_msg)
            self.user.send_notification(self.current_ip, 'error', e)


if __name__ == "__main__":
    IPChanger(sys.argv[1:])

Thank you in advance!

The full repository is available here with readme.

If you're interested, I also wrote a blog post on the process of writing the script: spoiler-alert: I tried semi-successfully to do the same thing with Selenium before I figured out the API!

\$\endgroup\$
4
\$\begingroup\$

"Current" directory

Your get_cwd is problematic for a list of reasons:

  • It does not get the current working directory; it sets the current working directory and returns that, which is a surprising side-effect given the method's name
  • It's not the current directory you're after; it's the directory of the source code
  • Spooky changes based on OS that try to make assumptions about what the current directory was before the method was called

This needs a serious re-think. It's only being used to initialize FILE_PATH, which in turn is being used as the base directory for USER_PICKLE and LOG_FILE. That reveals another issue: you really should not be putting log and user files next to source code. In a Unix-like directory tree, logs should go in /var/log, and your USER_PICKLE perhaps in /usr/share/my_app. The same vaguely applies in Windows: if your source is in Program Files, log and user files could go in either ProgramData of the root volume or the AppData directory for the service user.

Once you've chosen sane locations for your files, don't %-format in the base directories; use pathlib - particularly since you want to be portable, this will use the correct slashes.

Class caching

Reaching for pickle is a good decision. Mashing values into __dict__ like this:

        self.__dict__.update(self.load_user().__dict__)

perhaps less so. One consequence is that this will not be compatible with __slots__ if you go that route. Consider changing your __init__ to always construct a new instance, and keep your load_user as it is.

Finally exit?

Your finally clauses in IPChanger.__init__ are questionable. If I read it correctly, after you exit(5), you're still going to execute your finally block, which attempts a complete arg parse and application execution. The only thing stopping it would be that opts would have failed to populate.

This second finally:

        finally:
            sys.exit()

should just be deleted. The only thing I can imagine you intended here is to silence exceptions, but that's not a great idea.

Delete this:

    opts = []

and your first and second finally. Also consider moving your arg-parsing code into its own function since it's so lengthy.

Membership checks

Change these:

                elif opt in ('-e', '--email'):

into set literal checks:

                elif opt in {'-e', '--email'}:

Also, use this for response comparison:

if response in {'nohost', 'notfdqn'}:

Response parsing

I wasn't all that successful in finding documentation on what a response format from this API looks like, but one example I found on a random post (that I don't consider authoritative enough to link to) compares the entire response to "good" / "nochg" instead of a prefix. As such, is the slicing here:

        if response[:4] == 'good' or response[:5] == 'nochg':

actually necessary?

There are other aspects of your Requests usage that should be improved. You construct this URL:

f'https://{self.DNS_username}:{self.DNS_password}{self.BASE_URL}{self.domain}&myip='
 

in a method set_credentials that shouldn't include the update feature. It should also not include the &myip suffix.

Consider instead constructing this URL with urlunparse, rather than string formatting; leaving out the myip parameter, and later passing this as a key in a params dictionary to go to post.

Additionally, do not response.content.decode yourself, use response.text; and be sure you call raise_for_status to notice whether a call failed.

Security

This is a big no-no:

self.DNS_password = input("What's your autogenerated DNS password?: ")
    

It's vulnerable to over-the-shoulder password snooping. Use getpass instead.

\$\endgroup\$
4
  • \$\begingroup\$ I remember in being faster for small hard-coded tuples than hard-coded sets. Not sure if this is still the case for the current version of Python, though. \$\endgroup\$ – Graipher Jan 3 at 18:10
  • \$\begingroup\$ @Graipher AFAIK in Python > 3.2, if you use sets the peephole optimiser will replace that with a frozenset() object and membership tests against sets are O(1) constant operations. Testing against a tuple is O(n) worst case. Although testing against a set has to calculate the hash (higher constant cost, cached for immutable types), the cost for testing against a tuple other than the first element is always going to be higher. So on average, sets are easily faster even for small hard-coded tuples. \$\endgroup\$ – Grajdeanu Alex Jan 3 at 18:56
  • 1
    \$\begingroup\$ @GrajdeanuAlex Well, the reason I remember was that constructing a (small) tuple is so much faster than constructing a set that it made up for the slower in. I'll have to make some timings tomorrow to check if this is (still) the case... \$\endgroup\$ – Graipher Jan 3 at 19:29
  • 2
    \$\begingroup\$ @Reinderien thank you so much for all this; it's more detailed/useful than I could have hoped for \$\endgroup\$ – nihilok Jan 3 at 19:54

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.