Create auth frontend and filled out auth service to get user authenti…

…cation / authorization from github.

vro-streamlit/pyproject.toml

@@ -23,7 +23,6 @@ testpaths = [
 # Environment variables to use in pytests
 env = [
     "ENV=test-environment",
-    "DEBUG=True"
 ]
 
 [tool.coverage.run]

vro-streamlit/src/dev-requirements.txt

@@ -5,4 +5,6 @@ pytest>=7.4
 pytest-cov>=5.0
 pytest-env>=1.0
 pytest-mock>=3.11
+requests-mock>=1.10
 ruff>=0.6
+types-requests>=2.32

vro-streamlit/src/requirements.txt

@@ -1,3 +1,5 @@
 pandas==2.2.*
+requests==2.32.*
 starlette>=0.40.0
 streamlit==1.39.*
+validators==0.34.*

vro-streamlit/src/vro_streamlit/auth/auth_frontend.py

@@ -0,0 +1,59 @@
+import logging
+
+import streamlit as st
+
+import vro_streamlit.auth.auth_service as auth
+from vro_streamlit.auth.user import User
+
+AUTH_LOG_IN_BUTTON = 'auth_log_in_button'
+AUTH_LOG_OUT_BUTTON = 'auth_log_out_button'
+
+
+def log_out() -> None:
+    if st.button('Log Out', key=AUTH_LOG_OUT_BUTTON):
+        try:
+            auth.log_out(st.session_state.user)
+            st.success('Logged out successfully.')
+        except Exception as e:
+            logging.error(f'Failed to revoke token, but logged out anyways: {e}')
+
+        st.session_state.user = None
+        st.rerun()
+
+
+def log_in() -> None:
+    if st.button('Log In with GitHub', key=AUTH_LOG_IN_BUTTON, help='Log in to access more features.'):
+        # Step 1: Initiate the Device Flow
+        device_flow_data = auth.initiate_device_flow()
+
+        # Display the user code and verification URI
+        st.write(f'1. Go to [GitHub]({device_flow_data.verification_uri}) and enter the following code:')
+        st.code(device_flow_data.user_code)
+        st.write(f'2. This code expires in {device_flow_data.expires_in // 60} minutes.')
+
+        # Step 2: Poll for Access Token
+        with st.spinner('Waiting for authorization...'):
+            try:
+                access_token = auth.poll_for_token(device_flow_data.device_code, device_flow_data.interval)
+                try:
+                    # Step 3: Fetch User Info
+                    user_info = auth.fetch_user_info(access_token)
+                    st.success(f'Login successful! Logged in as: {user_info.login}')
+                    st.session_state.user = User(access_token, user_info.login, user_info.avatar_url)
+                    st.rerun()
+                except Exception as e:
+                    st.error(f'Could not gather user information: {e}')
+            except Exception as e:
+                st.error(f'Login failed: {e}')
+                st.session_state.user = None
+
+
+def show() -> None:
+    if not st.session_state.user:
+        log_in()
+    else:
+        log_out()
+
+
+if __name__ == '__main__':
+    show()

vro-streamlit/src/vro_streamlit/auth/auth_service.py

@@ -1,9 +1,78 @@
-from vro_streamlit.auth.user import User
+import time
 
+import requests
 
-def log_in() -> User:
-    return User('test')
+from vro_streamlit.auth.response_models import DeviceFlowResponse, UserInfoResponse
+from vro_streamlit.config import GITHUB_CLIENT_ID, GITHUB_CLIENT_SECRET
 
+# Replace with your actual values
+CLIENT_ID = GITHUB_CLIENT_ID
+CLIENT_SECRET = GITHUB_CLIENT_SECRET
 
-def log_out() -> bool:
-    return True
+DEVICE_CODE_URL = 'https://github.com/login/device/code'
+TOKEN_URL = 'https://github.com/login/oauth/access_token'
+USER_URL = 'https://api.github.com/user'
+REVOKE_URL = f'https://api.github.com/applications/{CLIENT_ID}/token'
+
+
+def initiate_device_flow() -> DeviceFlowResponse:
+    """Initiate the OAuth Device Flow and return a DeviceFlowResponse object."""
+    response = requests.post(
+        DEVICE_CODE_URL,
+        data={'client_id': CLIENT_ID, 'scope': 'user'},
+        headers={'Accept': 'application/json'},
+    )
+    response.raise_for_status()
+    data = response.json()
+    return DeviceFlowResponse(
+        data['device_code'],
+        data['user_code'],
+        data['verification_uri'],
+        int(data['expires_in']),
+        int(data['interval']),
+    )
+
+
+def poll_for_token(device_code: str, interval: int) -> str:
+    """Poll GitHub for an access token until the user authorizes the app."""
+    while True:
+        response = requests.post(
+            TOKEN_URL,
+            data={
+                'client_id': CLIENT_ID,
+                'device_code': device_code,
+                'grant_type': 'urn:ietf:params:oauth:grant-type:device_code',
+            },
+            headers={'Accept': 'application/json'},
+        )
+
+        data = response.json()
+        if 'access_token' in data:
+            return str(data['access_token'])
+        elif 'error' in data and data['error'] != 'authorization_pending':
+            raise Exception(f"Error: {data['error_description']}")
+
+        time.sleep(interval)
+
+
+def fetch_user_info(access_token: str) -> UserInfoResponse:
+    """Fetch user info using the access token."""
+    response = requests.get(
+        USER_URL,
+        headers={'Authorization': f'token {access_token}'},
+    )
+    response.raise_for_status()
+    data = response.json()
+    return UserInfoResponse(**data)
+
+
+def log_out(access_token: str) -> bool:
+    """Revoke the access token."""
+    response = requests.delete(
+        REVOKE_URL,
+        auth=(CLIENT_ID, CLIENT_SECRET),  # Basic Authentication
+        json={'access_token': access_token},  # Send token in the request body
+        headers={'Accept': 'application/vnd.github+json'},  # Recommended header for GitHub API
+    )
+    response.raise_for_status()
+    return response.status_code == 204

vro-streamlit/src/vro_streamlit/auth/response_models.py

@@ -0,0 +1,18 @@
+from typing import Any
+
+
+class DeviceFlowResponse:
+    def __init__(self, device_code: str, user_code: str, verification_uri: str, expires_in: int, interval: int):
+        self.device_code: str = device_code
+        self.user_code: str = user_code
+        self.verification_uri: str = verification_uri
+        self.expires_in: int = expires_in
+        self.interval: int = interval
+
+
+class UserInfoResponse:
+    def __init__(self, login: str, avatar_url: str, **kwargs: Any):
+        self.login = login
+        self.avatar_url = avatar_url
+        for key, value in kwargs.items():
+            setattr(self, key, value)

vro-streamlit/src/vro_streamlit/auth/user.py

@@ -1,3 +1,5 @@
 class User:
-    def __init__(self, username: str):
-        self.username: str = username
+    def __init__(self, access_token: str, username: str, avatar_url: str | None = None) -> None:
+        self.access_token = access_token
+        self.username = username
+        self.avatar_url = avatar_url

vro-streamlit/src/vro_streamlit/config.py

@@ -3,3 +3,5 @@
 
 ENV = getenv('ENV', 'local')
 DEBUG = bool(strtobool(getenv('DEBUG', 'False')))
+GITHUB_CLIENT_ID = getenv('GITHUB_CLIENT_ID', 'github_client_id')
+GITHUB_CLIENT_SECRET = getenv('GITHUB_CLIENT_SECRET', 'github_client_secret')

vro-streamlit/src/vro_streamlit/directory/home.py

@@ -1,34 +1,28 @@
 from importlib.resources import files
 
 import streamlit as st
+import validators
 
-import vro_streamlit.auth.auth_service as auth
+from vro_streamlit.auth import auth_frontend
 
-LOGIN_BUTTON = 'home_login_button'
 LOGO = files('vro_streamlit').joinpath('static/streamlit-logo.png').read_bytes()
 
 
-def update_login_status() -> None:
-    if not st.session_state.user:
-        st.session_state.user = auth.log_in()
-    else:
-        if auth.log_out():
-            st.session_state.user = None
-
-
 def show() -> None:
     col1, col2 = st.columns([0.04, 0.96])
     col1.image(LOGO, width=100)
     col2.header('Home')
     st.subheader('Welcome to the home page!')
 
-    user = st.session_state.get('user')
+    user = st.session_state.user
     if user:
-        st.write(f'Hello, {user.username}!')
-        st.button('Log Out', key=LOGIN_BUTTON, on_click=update_login_status)
+        if validators.url(user.avatar_url):
+            st.image(user.avatar_url, width=50)
+        st.write(f'Hello, **{user.username}**!')
     else:
         st.write('Please Log In')
-        st.button('Log In', key=LOGIN_BUTTON, on_click=update_login_status)
+
+    auth_frontend.show()
 
 
 if __name__ == '__main__':

vro-streamlit/src/vro_streamlit/main.py

@@ -1,48 +1,56 @@
+import logging
+
 import streamlit as st
 
 import vro_streamlit.auth.auth_service as auth
 import vro_streamlit.config as config
 import vro_streamlit.directory.home as home
+from vro_streamlit.config import DEBUG
 from vro_streamlit.directory.bie_events import claim_events, contention_events
 
-LOGIN_BUTTON = 'sidebar_login_button'
-LOGOUT_BUTTON = 'sidebar_logout_button'
+LOG_OUT_BUTTON = 'sidebar_log_out_button'
 
 st.set_page_config(page_title='VRO Streamlit', layout='wide')
+home_page = st.Page(home.show, title='Home', default=True)
+# BIE events
+bie_events = [
+    st.Page(claim_events.show, title='Claim Events', url_path='/claim_events'),
+    st.Page(contention_events.show, title='Contention Events', url_path='/contention_events'),
+]
+# examples
+examples = [
+    st.Page('directory/examples/text.py', title='Text'),
+    st.Page('directory/examples/dataframes.py', title='Dataframes'),
+    st.Page('directory/examples/water_quality.py', title='Water Quality'),
+]
 
 
 def init_session_state() -> None:
     st.session_state.setdefault('database_connected', True)
+    st.session_state.setdefault('logged_in', False)
     st.session_state.setdefault('user', None)
 
 
-def update_login_status() -> None:
-    if not st.session_state.user:
-        st.session_state.user = auth.log_in()
+def create_navigation() -> None:
+    if st.session_state.user:
+        nav = st.navigation({'Main': [home_page], 'BIE Events': bie_events, 'Examples': examples})
     else:
-        if auth.log_out():
-            st.session_state.user = None
+        nav = st.navigation({'Main': [home_page], 'Examples': examples})
+    nav.run()
 
 
-def create_navigation() -> None:
-    home_page = st.Page(home.show, title='Home', default=True)
-    # BIE events
-    bie_events = [
-        st.Page(claim_events.show, title='Claim Events', url_path='/claim_events'),
-        st.Page(contention_events.show, title='Contention Events', url_path='/contention_events'),
-    ]
-    # examples
-    examples = [
-        st.Page('directory/examples/text.py', title='Text'),
-        st.Page('directory/examples/dataframes.py', title='Dataframes'),
-        st.Page('directory/examples/water_quality.py', title='Water Quality'),
-    ]
-    nav = st.navigation({'Main': [home_page], 'BIE Events': bie_events, 'Examples': examples})
-    nav.run()
+def log_out() -> None:
+    try:
+        auth.log_out(st.session_state.user.access_token)
+        st.success('Logged out successfully.')
+    except Exception as e:
+        logging.error(f'Failed to revoke token, but logged out anyways: {e}')
+    st.session_state.user = None
 
 
 def create_sidebar() -> None:
     with st.sidebar:
+        user = st.session_state.user
         with st.container(border=True):
             col1, col2 = st.columns(2)
             with col1:
@@ -51,12 +59,18 @@ def create_sidebar() -> None:
                 st.markdown('Authorized', help='User authorization status')
             with col2:
                 st.markdown(f'`{config.ENV}`')
-                st.markdown(':large_green_circle:' if st.session_state.database_connected else ':red_circle:', unsafe_allow_html=True)
-                st.markdown(':large_green_circle:' if st.session_state.user else ':red_circle:', unsafe_allow_html=True)
+                st.markdown(':large_green_circle:' if st.session_state.database_connected else ':red_circle:')
+                st.markdown(':large_green_circle:' if user is not None else ':red_circle:')
+
+        if user is not None:
+            st.button('Log Out', use_container_width=True, on_click=log_out, key=LOG_OUT_BUTTON)
 
-        button_text = 'Log Out' if st.session_state.user else 'Log In'
-        button_key = LOGOUT_BUTTON if st.session_state.user else LOGIN_BUTTON
-        st.button(button_text, use_container_width=True, on_click=update_login_status, key=button_key)
+        if DEBUG:
+            with st.container(border=True):
+                st.write('Session State')
+                ss_dict = st.session_state.to_dict()
+                ss_dict['user'] = user.__dict__ if user else None
+                st.json(ss_dict)
 
 
 if __name__ == '__main__':

vro-streamlit/test/auth/test_auth_frontend.py

@@ -0,0 +1,54 @@
+from test.conftest import ACCESS_TOKEN, APP_TEST_TIMEOUT, AVATAR_URL, USERNAME
+from test.util import assert_markdown_contains_values
+
+import pytest
+from streamlit.testing.v1 import AppTest
+
+from vro_streamlit.auth import auth_frontend
+from vro_streamlit.auth.response_models import DeviceFlowResponse, UserInfoResponse
+from vro_streamlit.auth.user import User
+
+
+@pytest.fixture()
+def app_test():
+    app_test = AppTest.from_file('src/vro_streamlit/auth/auth_frontend.py', default_timeout=APP_TEST_TIMEOUT)
+    app_test.session_state.user = None
+    return app_test
+
+
+@pytest.fixture()
+def auth_service(mocker):
+    auth_service = mocker.patch('vro_streamlit.auth.auth_service')
+    auth_service.initiate_device_flow.return_value = DeviceFlowResponse('device_code', 'user_code', 'verification_uri', 600, 5)
+    auth_service.poll_for_token.return_value = ACCESS_TOKEN
+    auth_service.fetch_user_info.return_value = UserInfoResponse(login=USERNAME, avatar_url=AVATAR_URL)
+    auth_service.log_out.return_value = True
+    return auth_service
+
+
+def test_show_user_not_logged_in(app_test, auth_service) -> None:
+    app_test.run()
+    assert not app_test.exception
+
+    # Click log in button
+    app_test.button(key=auth_frontend.AUTH_LOG_IN_BUTTON).click().run()
+    assert not app_test.exception
+    assert_markdown_contains_values(app_test.code, 'user_code')
+
+    auth_service.initiate_device_flow.assert_called_once()
+    auth_service.poll_for_token.assert_called_once()
+    auth_service.fetch_user_info.assert_called_once()
+    app_test.session_state.user = User(ACCESS_TOKEN, USERNAME, AVATAR_URL)
+
+
+def test_show_user_logged_in(app_test, auth_service) -> None:
+    app_test.session_state.user = User(ACCESS_TOKEN, USERNAME, AVATAR_URL)
+    app_test.run()
+    assert not app_test.exception
+
+    # Click log out button
+    app_test.button(key=auth_frontend.AUTH_LOG_OUT_BUTTON).click().run()
+    assert not app_test.exception
+
+    auth_service.log_out.assert_called_once()
+    app_test.session_state.user = User(ACCESS_TOKEN, USERNAME, AVATAR_URL)