GraphAcademy

Building Neo4j Applications with Python

Project Introduction

The Neo4j Python Driver

Interacting with Neo4j

Project Backlog

Authenticating a User

At stage of the project, a user can register, but they are still unable to sign in. As with the previous Challenge, the authenticate() method is currently hard coded to accept only the email graphacademy@neo4j.com and password letmein.

In this challenge you will rewrite the authenticate() function of the AuthDAO to find the User node with the corresponding email and compare the password before issuing a JWT token.

But first, let’s take a look at how Authentication works in the application. If you prefer, you can skip straight to Implementing Authentication.

How Authentication Works

When a user attempts to access an API endpoint that requires authentication, the server checks for a JWT token.

When a user registers or signs in, a JWT token is generated and appended to the User record. This token is then stored by the UI and appended to Bearer to create an authorization header.

The token generation happens in the AuthDAO using the _generate_token() method.

When The API receives a request which includes the authorization header, the Flask-JWT library attempts to decode the value and makes the payload available by importing current_user from flask_jwt.

Login Route

When a user submits the login form on the website, a request is sent to http://localhost:3000/api/login with a username and password.

Implementing Authentication

To implement database authentication, you will modify the authenticate method in the AuthDAO.

python
api/dao/auth.py
def authenticate(self, email, plain_password):
    # TODO: Implement Login functionality
    if email == "graphacademy@neo4j.com" and plain_password == "letmein":
        # Build a set of claims
        payload = {
            "userId": "00000000-0000-0000-0000-000000000000",
            "email": email,
            "name": "GraphAcademy User",
        }

        # Generate Token
        payload["token"] = self._generate_token(payload)

        return payload
    else:
        return False

Your challenge is to update the authenticate() method to perform the following actions:

  1. Create a transaction function to find the User

  2. Execute the function within a Read Transaction

  3. Verify the User exists

  4. Compare Passwords

  5. Return User Details

Open api/dao/auth.py →

Create a transaction function to find the User

The Transaction function should be a simple query that uses Cypher to look up a :User node by the email parameter provided and return a single result.

python
Get User by Email
def get_user(tx, email):
    # Get the result
    result = tx.run("MATCH (u:User {email: $email}) RETURN u",
        email=email)

    # Expect a single row
    first = result.single()

    # No records? Return None
    if first is None:
        return None

    # Get the `u` value returned by the Cypher query
    user = first.get("u")

    return user

Execute the function within a Read Transaction

After opening up a new session, call the execute_read method to execute the get_user function above.

python
Call the Transaction Function
with self.driver.session() as session:
    user = session.execute_read(get_user, email=email)

Verify the User exists

If the user does not exist, then get_user will return None. In this case, return False.

python
# User not found, return False
if user is None:
    return False

Compare Passwords

The authenticate() method uses the hashpw function imported from bcrypt to encrypt the password. The library also provides a checkpw function for comparing a plain text value against the previously encrypted value.

If the check fails, return False.

python
# Passwords do not match, return false
if bcrypt.checkpw(plain_password.encode('utf-8'), user["password"].encode('utf-8')) is False:
    return False

Return User Details

Finally, if the user exists and the password comparison returns true, generate a JWT token and return it along with information about the User.

python
# Generate JWT Token
payload = {
    "userId": user["userId"],
    "email":  user["email"],
    "name":  user["name"],
}

payload["token"] = self._generate_token(payload)

return payload

Once you have applied these changes to the authenticate() method, scroll to Testing to verify that the method works as expected.

Working Solution

Click here to reveal the completed authenticate() method:
python
api/dao/auth.py
def authenticate(self, email, plain_password):
    def get_user(tx, email):
        # Get the result
        result = tx.run("MATCH (u:User {email: $email}) RETURN u",
            email=email)

        # Expect a single row
        first = result.single()

        # No records? Return None
        if first is None:
            return None

        # Get the `u` value returned by the Cypher query
        user = first.get("u")

        return user

    with self.driver.session() as session:
        user = session.execute_read(get_user, email=email)

        # User not found, return False
        if user is None:
            return False

        # Passwords do not match, return false
        if bcrypt.checkpw(plain_password.encode('utf-8'), user["password"].encode('utf-8')) is False:
            return False

        # Generate JWT Token
        payload = {
            "userId": user["userId"],
            "email":  user["email"],
            "name":  user["name"],
        }

        payload["token"] = self._generate_token(payload)

        return payload

Testing

To test that this functionality has been correctly implemented, run the following code in a new terminal session:

sh
Running the test
pytest tests/05_authentication__test.py

The test file is located at tests/05_authentication__test.py.

Are you stuck? Click here for help

If you get stuck, you can see a working solution by checking out the 05-authentication branch by running:

sh
Check out the 05-authentication branch
git checkout 05-authentication

You may have to commit or stash your changes before checking out this branch. You can also click here to expand the Support pane.

Verifying the Test

This test creates a new :User node in the database with an email address of authenticated@neo4j.com using the register() method on the AuthDAO and then attempts to verify the user using the authenticate method.

Hit the Check Database button below to verify that the test has been successfully run.

Hint

At the end of the test, a piece of code finds the User node and sets the authenticatedAt property to the current date and time using the Cypher datetime() function. As long as this value is within the past 24 hours, the test should pass.

You can run the following query to check for the user within the database. If the shouldVerify value returns true, the verification should be successful.

cypher
MATCH (u:User {email: 'authenticated@neo4j.com'})
RETURN u.email, u.authenticatedAt,
    u.authenticatedAt >= datetime() - duration('PT24H') AS shouldVerify

Solution

The following statement will mimic the behaviour of the test, merging a new :User node with the email address authenticated@neo4j.com, assigning a random UUID value to the .userId property and setting an .authenticatedAt property to the current date and time.

cypher
MERGE (u:User {email: "authenticated@neo4j.com"})
SET u.authenticatedAt = datetime()

Once you have run this statement, click Try again…​ to complete the challenge.

Lesson Summary

In this Challenge, you have updated the AuthDAO to authenticate a User using the data held in the Sandbox database.

In the next Challenge, you will save user ratings to the database.