fireauth.py

"""
The Firebase Auth REST SDK.

## Quickstart

The only thing to get going is to call `fireauth.init(API_KEY)`, where
`API_KEY` refers to the Web API Key, which can be obtained on the
[project settings](https://console.firebase.google.com/project/_/settings/general/?authuser=0)
page in your admin console. When not passed the `API_KEY`, the `API_KEY`
is picked up from the `FIREBASE_API_KEY` environment variable.

    import fireauth
    client = fireauth.init()

This initializes the default client which can be used to talk to your
firebase backend.
"""

__version__ = "0.1.0"
__author__ = "Danilo Braband <dbraband@gmail.com>"
__url__ = "https://github.com/dabio/fireauth"
__license__ = "MIT"

import gzip
import io
import json
import logging
import os
import urllib.error
import urllib.request
from collections import namedtuple

__all__ = ("init", "Client", "MissingAPIKeyException")

User = namedtuple("User", "user_id,email,token_id,refresh_token,expires_at")
User.__doc__ = """
Container class that holds all user elements that are necessary for working
with the Firebase API.
"""


def init(*args, **kwargs):
    """
    Initializes the SDK client.

    This takes the same arguments as the client constructor.
    """
    client = Client(*args, **kwargs)

    return client


class MissingAPIKeyException(Exception):
    "Exception to be thrown when no api key is provided."


class APIConnectionException(Exception):
    "Exception to be thrown when no connection to remote servers can be made."


class APIInvalidResponseException(Exception):
    "Error response from the API."

    def __init__(self, code: int, reason: str):
        super().__init__(f"Response error ({code}): {reason}")


class FirebaseResponse:
    "Response object with attributes of a Firebase API response."

    def __init__(self, body, code, headers):
        self.code = code
        self.headers = headers
        self.body = body
        self.data = json.loads(body)


class Client:
    """
    The client is responsible for calling the Firebase REST API. It
    takes an api_key as argument.
    """

    def __init__(self, api_key=None):
        self.api_key = api_key or os.getenv("FIREBASE_API_KEY")
        if self.api_key is None:
            raise MissingAPIKeyException("No API Key defined.")
        self.log = logging.getLogger(__name__)
        self.log.setLevel(logging.DEBUG)

    def _call_firebase(self, method: str, data: dict):
        url = f"{API_URL}{method}?key={self.api_key}"
        body, code, headers = _do_request("POST", url, data)
        response = self._interpret_response(body, code, headers)

    def _interpret_response(self, body, code, headers):
        if hasattr(body, "decode"):
            body = body.decode("utf-8")
        try:
            response = FirebaseResponse(body, code, headers)
        except json.JSONDecodeError as err:
            _handle_request_error(err)

        if not (200 <= code < 300):
            # {'error': {'code': 400, 'errors': [...], 'message': 'EMAIL_EXISTS'}}
            raise APIInvalidResponseException(
                code, response.body.get("error", {}).get("message")
            )

        return response

    def sign_up(self, email: str, password: str):
        """
        Create a new email and password user.
        """
        body = {
            "email": email,
            "password": password,
            "returnSecureToken": True,
        }
        return self._call_firebase("signupNewUser", body)

    def sign_in(self, email: str, password: str):
        """
        Sign in a user with email and password. Uses the sign_up method.
        """

    def change_email(self, user: User, email: str):
        """
        Change user's email.
        """

    def change_password(self, user: User, password: str):
        """
        Change user's password.
        """

    def send_password_reset(self, email: str) -> bool:
        """
        Send password reset for the given email.
        """

    def verify_password_reset_code(self, code: str) -> User:
        """
        Verify password reset code. Only the email attribute will be filled in
        returned user object.
        """

    def confirm_password_reset(self, code: str, password: str) -> User:
        """
        Apply a password reset change. Only email attribute will be filled in
        returned User object.
        """

    def delete_account(self, user: User) -> bool:
        """
        Delete the given user object.
        """


def _handle_request_error(err):
    msg = "Unexpected error communicating with Firebase."
    raise APIConnectionException(msg)


def _do_request(method: str, url: str, data: dict):
    """
    Call the Firebase Auth REST API as documented here:
    https://firebase.google.com/docs/reference/rest/auth/?authuser=0
    """
    body = io.BytesIO()
    with gzip.GzipFile(fileobj=body, mode="w") as file:
        file.write(json.dumps(data, allow_nan=False).encode("utf-8"))

    headers = {"Content-Type": "application/json", "Content-Encoding": "gzip"}
    req = urllib.request.Request(url, body.getvalue(), headers, method=method)
    try:
        with urllib.request.urlopen(
            req, timeout=API_TIMEOUT
        ) as res:  # type: http.client.HTTPResponse
            code = res.code
            body = res
            headers = dict(res.info())
    except urllib.error.HTTPError as res:  # type: urllib.error.HTTPError
        code = res.code
        body = res
        headers = dict(res.info())
    except (urllib.error.URLError, ValueError) as err:
        _handle_request_error(err)
    lower_headers = dict((k.lower(), v) for k, v in headers.items())

    return body, code, lower_headers


# Constants and globals

API_URL = "https://www.googleapis.com/identitytoolkit/v3/relyingparty/"
API_TIMEOUT = 10  # in seconds

test_client.py

import pytest

import fireauth


def test_api_key_env(monkeypatch):
    """
    Checks if the client stores the api key when in environment variable.
    """
    api_key = "foobar"
    monkeypatch.setenv("FIREBASE_API_KEY", api_key)
    client = fireauth.init()
    assert client.api_key == api_key


def test_api_key_args():
    """
    Checks wether the api key can be passed as a parameter to the init
    function.
    """
    api_key = "foobar"
    client = fireauth.init(api_key)
    assert client.api_key == api_key


def test_api_key_not_set(monkeypatch):
    """
    Checks if the module fires an error when no API_KEY was given and no
    environment variable was set.
    """
    monkeypatch.delenv("FIREBASE_API_KEY")
    with pytest.raises(fireauth.MissingAPIKeyException):
        fireauth.init()


def test_sign_up():
    """
    Checks if a user is able to sign up. the signed up user will be created at
    the end of the test.
    """
    client = fireauth.init()
    email = "test@signup.com"
    user = client.sign_up(email, "password")
    assert user.email == email
    assert client.delete_account(user)


def test_double_sign_up():
    """
    Checks for sign up tries with duplicate email address.
    """
    client = fireauth.init()
    email = "test@double.signup"
    user = client.sign_up(email, "password")
    assert user.email == email
    with pytest.raises(fireauth.exceptions.APIException):
        client.sign_up(email, "password")
    assert client.delete_account(user)


def test_sign_in():
    """
    Checks if the user can sign in with the given credentials. A new user
    object will be created before deleting it again.
    """
    client = fireauth.init()
    email = "test@signin.com"
    password = "password"
    user = client.sign_up(email, password)
    assert user.email == email
    login_user = client.sign_in(email, password)
    assert user.user_id == login_user.user_id
    assert client.delete_account(user)


def test_send_password_reset():
    """
    Checks password reset.
    """
    client = fireauth.init()
    client.send_password_reset("pass@word.com")