dynuapi/dynuapi.py

import os
import requests
import json
import pdb

###
# Dict helper class.
# Defined at top level so it can be pickled.
###
class AttribAccessDict(dict):
    def __getattr__(self, attr):
        if attr in self:
            return self[attr]
        else:
            raise AttributeError("Attribute not found: " + str(attr))

    def __setattr__(self, attr, val):
        if attr in self:
            raise AttributeError("Attribute-style access is read only")
        super(AttribAccessDict, self).__setattr__(attr, val)

class Dynu:

    name = 'Python wrapper for Dynu {d}DNS service'

    def __init__(self, api_base_url=None, api_key=None, session=None):

        self.file_path = "secrets/secrets.txt"

        is_setup = self.__check_setup(self)

        if is_setup:

            self.api_base_url = self.__get_parameter("api_base_url", self.file_path)
            self.api_key = self.__get_parameter("api_key", self.file_path)

        else:

            self.api_base_url, self.api_key = self.__setup(self)

        if session:
            self.session = session
        else:
            self.session = requests.Session()

    def dns(self):

        endpoint = self.api_base_url + '/dns'

        headers = {
                "Accept": "application/json",
                "API-Key": self.api_key
                }

        response = self.__api_request('GET', endpoint, headers=headers)

        response = self.__json_allow_dict_attrs(response.json())

        return response

    def update(self, id, name, group, ipv4Address, ipv6Address=None, ttl=90, ipv4=True, ipv6=None, ipv4WildcardAlias=None, ipv6WildcardAlias=None, allowZoneTransfer=None, dnssec=None):

        endpoint = self.api_base_url + '/dns/{0}'.format(id)

        headers = {
                "Accept": "application/json",
                "API-Key": self.api_key
                }

        data = {'name':name,
            'group':group,
            'ipv4Address':ipv4Address,
            'ipv6Address':'',
            'ttl': ttl,
            'ipv4':True,
            'ipv6':False,
            'ipv4WildcardAlias':True,
            'ipv6WildcardAlias':False,
            'allowZoneTransfer':False,
            "dnssec":False
        }

        payload = json.dumps(data)

        response = self.__api_request('POST', endpoint, headers=headers, data=payload)

        response = self.__json_allow_dict_attrs(response.json())

        return response


    @staticmethod
    def __check_setup(self):

        is_setup = False

        if not os.path.isfile(self.file_path):
            print(f"File {self.file_path} not found, running setup.")
        else:
            is_setup = True

        return is_setup

    @staticmethod
    def __setup(self):

        if not os.path.exists('secrets'):
            os.makedirs('secrets')

        self.api_base_url = input("Dynu {d}DNS API base url, in ex. 'https://api.dynu.com/v2': ")
        self.api_key = input("Dynu {d}DNS API key: ")

        if not os.path.exists(self.file_path):
            with open(self.file_path, 'w'): pass
            print(f"{self.file_path} created!")

        with open(self.file_path, 'a') as the_file:
            print("Writing Dynu parameters to " + self.file_path)
            the_file.write(f'api_base_url: {self.api_base_url}\n'+f'api_key: {self.api_key}\n')

        return (self.api_base_url, self.api_key)

    @staticmethod
    def __get_parameter(parameter, file_path ):

        with open( file_path ) as f:
            for line in f:
                if line.startswith( parameter ):
                    return line.replace(parameter + ":", "").strip()

        print(f'{file_path} Missing parameter {parameter}')
        sys.exit(0)

    def __api_request(self, method, endpoint, headers={}, data={}):

        response = None

        try:

            kwargs = dict(headers=headers, data=data)

            response = self.session.request(method, url = endpoint, **kwargs)

        except Exception as e:

            raise DynuNetworkError(f"Could not complete request: {e}")

        if response is None:

            raise DynuIllegalArgumentError("Illegal request.")

        if not response.ok:

                try:
                    if isinstance(response, dict) and 'error' in response:
                        error_msg = response['error']
                    elif isinstance(response, str):
                        error_msg = response
                    else:
                        error_msg = None
                except ValueError:
                    error_msg = None

                if response.status_code == 404:
                    ex_type = DynuNotFoundError
                    if not error_msg:
                        error_msg = 'Endpoint not found.'
                        # this is for compatibility with older versions
                        # which raised DynuAPIError('Endpoint not found.')
                        # on any 404
                elif response.status_code == 401:
                    ex_type = DynuUnauthorizedError
                elif response.status_code == 422:
                    return response
                elif response.status_code == 500:
                    ex_type = DynuInternalServerError
                elif response.status_code == 502:
                    ex_type = DynuBadGatewayError
                elif response.status_code == 503:
                    ex_type = DynuServiceUnavailableError
                elif response.status_code == 504:
                    ex_type = DynuGatewayTimeoutError
                elif response.status_code >= 500 and \
                    response.status_code <= 511:
                    ex_type = DynuServerError
                else:
                    ex_type = DynuAPIError

                raise ex_type(
                    'Dynu API returned error',
                    response.status_code,
                    response.reason,
                    error_msg)

        else:

            return response

    @staticmethod
    def __json_allow_dict_attrs(json_object):
        """
        Makes it possible to use attribute notation to access a dicts
        elements, while still allowing the dict to act as a dict.
        """
        if isinstance(json_object, dict):
            return AttribAccessDict(json_object)
        return json_objecte

##
# Exceptions
##
class DynuError(Exception):
    """Base class for Dynu.py exceptions"""

class DynuIOError(IOError, DynuError):
    """Base class for Dynu.py I/O errors"""

class DynuNetworkError(DynuIOError):
    """Raised when network communication with the server fails"""
    pass
class DynuAPIError(DynuError):
    """Raised when the gitea API generates a response that cannot be handled"""
    pass
class DynuServerError(DynuAPIError):
    """Raised if the Server is malconfigured and returns a 5xx error code"""
    pass
class DynuInternalServerError(DynuServerError):
    """Raised if the Server returns a 500 error"""
    pass

class DynuBadGatewayError(DynuServerError):
    """Raised if the Server returns a 502 error"""
    pass

class DynuServiceUnavailableError(DynuServerError):
    """Raised if the Server returns a 503 error"""
    pass
class DynuGatewayTimeoutError(DynuServerError):
    """Raised if the Server returns a 504 error"""
    pass
class DynuNotFoundError(DynuAPIError):
    """Raised when the gitea API returns a 404 Not Found error"""
    pass

class DynuUnauthorizedError(DynuAPIError):
    """Raised when the gitea API returns a 401 Unauthorized error

       This happens when an OAuth token is invalid or has been revoked,
       or when trying to access an endpoint that can't be used without
       authentication without providing credentials."""
    pass