Merge pull request #192 from CodeForPhilly/user_api_testing

New secrets, new user api tests, prefixed usernames with base_

Author: c-simpson

src/server/Dockerfile

@@ -41,4 +41,7 @@ RUN set FLASK_APP=server/app.py
 RUN export FLASK_APP
 
 # This abomination ensures that the PG server has finished its restart cycle
-CMD echo "SLEEPING 10"; sleep 10; echo "WAKING"; alembic upgrade head ; python -m flask run --host=0.0.0.0
+CMD echo "SLEEPING 10"; sleep 10; echo "WAKING"; alembic upgrade head ; python -m flask run --host=0.0.0.0 --no-reload
+
+# --no-reload prevents Flask restart, which usually happens in middle of create_base_users()
+#TODO: SECURITY - ensure we are not running in debug mode in production

src/server/app.py

@@ -4,16 +4,17 @@
 
 from flask_jwt_extended import JWTManager
 
+from secrets import JWT_SECRET, APP_SECRET_KEY
 
 app = Flask(__name__)
 
-app.config["JWT_SECRET_KEY"] = "super-secret"  # TODO: SECURITY  Change this!
+app.config["JWT_SECRET_KEY"] = JWT_SECRET  
 app.config["JWT_ACCESS_TOKEN_EXPIRES"] = 300  # Seconds for timeout. 60 for testing.
 jwt = JWTManager(app)
 
 
 # def create_app():
-app.secret_key = "1u9L#*&I3Ntc"  # TODO: SECURITY  Change this!
+app.secret_key = APP_SECRET_KEY  
 app.config["MAX_CONTENT_LENGTH"] = 500 * 1024 * 1024  # 500 Megs
 app.config["SEND_FILE_MAX_AGE_DEFAULT"] = 0
 from api.admin_api import admin_api

src/server/test_api.py

@@ -1,5 +1,9 @@
 import pytest, socket, requests, os
 
+from secrets import BASEUSER_PW, BASEADMIN_PW
+
+jwt_token = ''
+
 #
 # Run 'pytest' from the command line
 #
@@ -21,43 +25,170 @@
     SERVER_URL = "http://server:5000"
     IS_LOCAL = False
 
-###  DNS lookup tests
+###  DNS lookup tests  ##############################
 
-# Ensure DNS not resolving bad host names
 def test_bad_dns():
+    """Verify DNS not resolving bad host names."""
     with pytest.raises(socket.gaierror):
         socket.getaddrinfo("bad_server_name_that_should_not_resolve", "5000")
 
 
-# Do we get IPs for good names?
-
-
 @pytest.mark.skipif(IS_LOCAL, reason="Not run when IS_LOCAL")
 def test_db_dns():
-    assert (
-        len(socket.getaddrinfo("db", "5000")) > 0
-    )  # getaddrinfo works for IPv4 and v6
+    """Verify we get IP for DB server."""
+
+    # getaddrinfo works for IPv4 and v6
+    try:
+        gai = socket.getaddrinfo("db", "5000")
+    except:
+        pytest.fail('getaddrinfo() failed for db', pytrace=False)
+
+    assert len(gai) > 0
 
 
 @pytest.mark.skipif(IS_LOCAL, reason="Not run when IS_LOCAL")
 def test_server_dns():
-    assert len(socket.getaddrinfo("server", "5000")) > 0
+    """Verify we get IP for API server."""
+    try:
+        gai = socket.getaddrinfo("server", "5000")
+    except socket.gaierror:
+        pytest.fail('getaddrinfo() failed for server', pytrace=False)
+
+    assert len(gai) > 0
 
 
 @pytest.mark.skipif(IS_LOCAL, reason="Not run when IS_LOCAL")
 def test_client_dns():
-    assert len(socket.getaddrinfo("client", "5000")) > 0
+    """Verify we get IP for client."""
+    try:
+        gai = socket.getaddrinfo("client", "5000")
+    except socket.gaierror:
+        pytest.fail('getaddrinfo() failed for client', pytrace=False)
 
+    assert len(gai) > 0
 
-# Simple API tests
+# Simple API tests  ################################################
 
 
 def test_currentFiles():
+    """360 view Current Files list"""
     response = requests.get(SERVER_URL + "/api/listCurrentFiles")
     assert response.status_code == 200
 
 
 def test_statistics():
+    """360 view Statistics"""
     response = requests.get(SERVER_URL + "/api/statistics")
     assert response.status_code == 200
 
+
+def test_usertest():
+    """Verify liveness test works"""
+    response = requests.get(SERVER_URL + "/api/user/test")
+    assert response.status_code == 200
+
+########   Dependent tests   #################################
+
+#      Store info across tests
+class State:
+    def __init__(self):
+        self.state = {}
+
+@pytest.fixture(scope='session')
+def state() -> State:
+    state = State()
+    state.state['from_fixture'] = 0
+    return state
+
+
+def test_userlogin(state: State):
+    """Verify base_user can log in/get JWT."""
+    data = {"username":"base_user", "password" : BASEUSER_PW}
+
+    response = requests.post(SERVER_URL + "/api/user/login_json", json=data)
+    assert response.status_code == 200
+
+    try:
+        jwt_token = response.json()['access_token']
+    except:
+        pytest.fail('Did not get access token',  pytrace=False)
+
+    assert len(jwt_token) > 16
+
+    # Store the token for later use
+    state.state['base_user'] = jwt_token
+
+
+def test_useraccess(state: State):
+    """Verify logged-in base_user can use JWT to access test_auth"""
+    # Build auth string value including token from state
+    b_string = 'Bearer ' + state.state['base_user']
+
+    assert len(b_string) > 24
+
+    auth_hdr = {'Authorization' : b_string}
+    response = requests.get(SERVER_URL + "/api/user/test_auth", headers=auth_hdr)
+    assert response.status_code == 200
+
+
+def test_user_bad_pw():
+    """Verify base_user with bad pw fails"""
+    data = {"username":"base_user", "password" : 'some_bad_password'}
+
+    response = requests.post(SERVER_URL + "/api/user/login_json", json=data)
+    assert response.status_code == 401
+
+
+def test_inact_userblocked(state: State):
+    """Verify base_user_inact can't login because marked inactive."""
+    # Same pw as base_user
+    data = {"username":"base_user_inact", "password" : BASEUSER_PW}
+    response = requests.post(SERVER_URL + "/api/user/login_json", json=data)
+    assert response.status_code == 401
+
+
+###   Admin-level tests ######################################
+
+def test_adminlogin(state: State):
+    """Verify base_admin can log in/get JWT."""
+    data = {"username":"base_admin", "password" : BASEADMIN_PW}
+
+    response = requests.post(SERVER_URL + "/api/user/login_json", json=data)
+    assert response.status_code == 200
+
+    try:
+        jwt_token = response.json()['access_token']
+    except:
+        pytest.fail('Did not get access token',  pytrace=False)
+
+    assert len(jwt_token) > 16
+
+    # Store the token for later use
+    state.state['base_admin'] = jwt_token
+
+
+def test_admingetusers(state: State):
+    """Verify logged-in base_admin can use JWT to get user list """
+    # Build auth string value including token from state
+    b_string = 'Bearer ' + state.state['base_admin']
+
+    assert len(b_string) > 24
+
+    auth_hdr = {'Authorization' : b_string}
+    response = requests.get(SERVER_URL + "/api/admin/user/get_users", headers=auth_hdr)
+    assert response.status_code == 200
+
+    userlist = response.json()
+    assert len(userlist) > 1
+
+
+def test_usergetusers(state: State):
+    """Verify logged-in base_user *cannot* use JWT to get user list """
+    # Build auth string value including token from state
+    b_string = 'Bearer ' + state.state['base_user']
+
+    assert len(b_string) > 24
+
+    auth_hdr = {'Authorization' : b_string}
+    response = requests.get(SERVER_URL + "/api/admin/user/get_users", headers=auth_hdr)
+    assert response.status_code == 403

src/server/user_mgmt/base_users.py

@@ -3,6 +3,9 @@
 from api import user_api
 import sqlalchemy as sa
 
+from secrets import BASEUSER_PW, BASEEDITOR_PW, BASEADMIN_PW
+
+
 from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
 
 metadata = sa.MetaData()
@@ -35,30 +38,30 @@ def create_base_users():  # TODO: Just call create_user for each
             pu = sa.Table("pdp_users", metadata, autoload=True, autoload_with=engine)
 
             # user
-            pw_hash = user_api.hash_password("userpw")
+            pw_hash = user_api.hash_password(BASEUSER_PW)
             ins_stmt = pu.insert().values(
-                username="user", password=pw_hash, active="Y", role=1,
+                username="base_user", password=pw_hash, active="Y", role=1,
             )
             connection.execute(ins_stmt)
 
             # INactive user
             # Reuse pw hash
             ins_stmt = pu.insert().values(
-                username="user_inact", password=pw_hash, active="N", role=1,
+                username="base_user_inact", password=pw_hash, active="N", role=1,
             )
             connection.execute(ins_stmt)
 
             # editor
-            pw_hash = user_api.hash_password("editorpw")
+            pw_hash = user_api.hash_password(BASEEDITOR_PW)
             ins_stmt = pu.insert().values(
-                username="editor", password=pw_hash, active="Y", role=2,
+                username="base_editor", password=pw_hash, active="Y", role=2,
             )
             connection.execute(ins_stmt)
 
             # admin
-            pw_hash = user_api.hash_password("adminpw")
+            pw_hash = user_api.hash_password(BASEADMIN_PW)
             ins_stmt = pu.insert().values(
-                username="admin", password=pw_hash, active="Y", role=9,
+                username="base_admin", password=pw_hash, active="Y", role=9,
             )
             connection.execute(ins_stmt)