<filename>__init__.py
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8

"""
Meerschaum plugin to add SSO login functionality to the API.
"""

__version__ = '0.0.1'
required = ['authlib', 'httpx', 'sqlalchemy',]
from meerschaum.plugins import api_plugin
from typing import Union, Optional

@api_plugin
def init(app):
    import json
    import hashlib
    from urllib.parse import quote
    from fastapi import FastAPI, Response, Cookie, Header
    import httpx
    import sqlalchemy
    from starlette.config import Config
    from starlette.requests import Request
    from starlette.middleware.sessions import SessionMiddleware
    from starlette.responses import HTMLResponse, RedirectResponse
    from authlib.integrations.starlette_client import OAuth, OAuthError
    from .config import get_sso_config, GOOGLE_CONF_URL, FACEBOOK_CONF_DICT
    from meerschaum import get_connector
    from meerschaum.api import get_uvicorn_config
    from meerschaum.connectors.sql._tools import get_sqlalchemy_table

    canonical_host = get_sso_config('canonical_hostname')
    prepend_path = get_sso_config('prepend')
    google_client_id = get_sso_config('google', 'id')
    google_client_secret = get_sso_config('google', 'secret')
    google_callback_slug = get_sso_config('google', 'callback')
    facebook_app_id = get_sso_config('facebook', 'id')
    facebook_app_secret = get_sso_config('facebook', 'secret')
    facebook_callback_slug = get_sso_config('facebook', 'callback')
    db_label = get_sso_config('db_label')
    conn = get_connector('sql', db_label)
    tables = {
        'people': get_sqlalchemy_table('people', connector=conn),
        'people-with-contact-info': get_sqlalchemy_table(
            'people-with-contact-info', connector=conn
        ),
        'session': get_sqlalchemy_table('session', connector=conn),
        'contact_info': get_sqlalchemy_table('contact_info', connector=conn),
    }

    oauth = OAuth()
    oauth.register(
        name='google',
        client_id=google_client_id,
        client_secret=google_client_secret,
        server_metadata_url=GOOGLE_CONF_URL,
        client_kwargs={
            'scope': 'openid email profile',
        },
    )
    oauth.register(
        name='facebook',
        client_id=facebook_app_id,
        client_secret=facebook_app_secret,
        redirect_uri=facebook_callback_slug,
        client_kwargs={
            'scope': 'email',
        },
        **FACEBOOK_CONF_DICT
    )
    app.add_middleware(SessionMiddleware, secret_key=google_client_secret)

    sso_google_path = prepend_path + '/sso/google'
    sso_facebook_path = prepend_path + '/sso/facebook'
    redirect_target = '/'

    def get_address_from_email(email: str) -> Union[str, None]:
        """
        Return a physical address linked to an email address.
        """
        #  q = f"SELECT address FROM \"people-with-contact-info\" WHERE email = '{email}'"
        t = tables['people-with-contact-info']
        q = sqlalchemy.select([t.c.address]).where(t.c.email == email)
        return conn.value(q)

    def get_email_from_name(name: str) -> Union[str, None]:
        """
        Return an email address from a real name.
        """
        fname = name.split(' ')[0]
        lname = ' '.join(name.split(' ')[1:])
        #  q = (
            #  "SELECT first, last, id FROM people "
            #  + f"WHERE first = '{fname}' AND last = '{lname}'"
        #  )
        t = tables['people']
        q = sqlalchemy.select([t.c.first, t.c.last, t.c.id]).where(
            t.c.first == fname and t.c.last = lname
        )
        print(str(q))

        if conn.value(q) is None or True:
            fakeemail = fname + lname + '@<EMAIL>'
            return fakeemail

        return conn.value(q)

    def get_host_from_referer(referer: Optional[str]) -> str:
        """
        Parse the host section of a url from a full url referer.
        """
        if referer is None:
            return canonical_host
        redirect_full_uri = ':'.join(str(referer).split(':')[0:2]) 
        redirect_base_uri = '/'.join(redirect_full_uri.split('/')[0:3])
        return redirect_base_uri

    def record_login_hash(name: str, email: str):
        """
        Record the current login checksum to the database for validation calls later.
        """
        pattern = f'{quote(name)}TO{quote(email)}INTO{canonical_host}'
        print(pattern)
        result = hashlib.md5(pattern.encode()).hexdigest()

        t = tables['session']
        q = sqlalchemy.select([t.c.login_email]).where(t.c.key == result)
        login_exists = conn.value(q) is not None
        if not login_exists:
            q = sqlalchemy.insert(t).values({'key': result, 'login_email': email})
            conn.exec(q)
            print(f'inserted key for user at {email}')

        return result

    @app.get(prepend_path)
    async def testConnection():
        return "<h1>SSO is alive</h1>"

    @app.get(sso_google_path)
    async def google_login(request: Request, referer: Optional[str] = Header(None)):
        redirect_base = get_host_from_referer(referer)
        redirect_uri = redirect_base + google_callback_slug
        if redirect_base.find('staging') != -1:
            print(str(referer) + ' -> ' + redirect_uri)
        return await oauth.google.authorize_redirect(request, redirect_uri)

    @app.get(sso_google_path + '/callback')
    async def google_callback(request: Request):
        token = await oauth.google.authorize_access_token(request)
        user = await oauth.google.parse_id_token(request, token)

        login_token = record_login_hash(user.name, user.email)

        response = RedirectResponse(redirect_target)
        response.set_cookie(key='login_provider', value='google', secure=True)
        response.set_cookie(key='login_name', value=quote(user.name), secure=True)
        response.set_cookie(key='login_email', value=quote(user.email), secure=True)
        response.set_cookie(key='login_id', value=login_token, secure=True)
        response.set_cookie(key='login_name_verified', value=True, secure=True)
        response.set_cookie(key='login_email_verified', value=True, secure=True)
        return response

    @app.get(sso_facebook_path)
    async def facebook_login(request: Request, referer: Optional[str] = Header(None)):
        redirect_base = get_host_from_referer(referer)
        redirect_uri = redirect_base + facebook_callback_slug
        if redirect_base.find('staging') != -1:
            print(str(referer) + ' -> ' + redirect_uri)
        return await oauth.facebook.authorize_redirect(request, redirect_uri)

    @app.get(sso_facebook_path + '/callback')
    async def facebook_callback(request: Request):
        token = await oauth.facebook.authorize_access_token(request)
        access_token = token.get('access_token', None)
        async with httpx.AsyncClient() as client:
            r = await client.get(
                'https://graph.facebook.com/me?fields=id,name,email',
                params={'access_token': access_token}
            )
        name_verified = True
        email_verified = False

        user_dict = r.json()
        name = user_dict.get('name', None)
        ### TODO handle missing name
        email = user_dict.get('email', None)
        if email == None:
            email = get_email_from_name(name) 
        else:
            email_verified = True
        ### TODO handle missing email

        login_token = record_login_hash(name, email if email != None else 'FB-NO-EMAIL')

        response = RedirectResponse(redirect_target)
        response.set_cookie(key='login_provider', value='facebook', secure=True)
        response.set_cookie(key='login_name', value=quote(name), secure=True)
        response.set_cookie(key='login_email', value=quote(email), secure=True)
        response.set_cookie(key='login_id', value=login_token, secure=True)
        response.set_cookie(key='login_name_verified', value=str(name_verified), secure=True)
        response.set_cookie(key='login_email_verified', value=str(email_verified), secure=True)
        return response


<filename>__init__.py
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8

"""
Meerschaum plugin to add SSO login functionality to the API.
"""

__version__ = '0.0.1'
required = ['authlib', 'httpx', 'sqlalchemy',]
from meerschaum.plugins import api_plugin
from typing import Union, Optional

@api_plugin
def init(app):
    import json
    import hashlib
    from urllib.parse import quote
    from fastapi import FastAPI, Response, Cookie, Header
    import httpx
    import sqlalchemy
    from starlette.config import Config
    from starlette.requests import Request
    from starlette.middleware.sessions import SessionMiddleware
    from starlette.responses import HTMLResponse, RedirectResponse
    from authlib.integrations.starlette_client import OAuth, OAuthError
    from .config import get_sso_config, GOOGLE_CONF_URL, FACEBOOK_CONF_DICT
    from meerschaum import get_connector
    from meerschaum.api import get_uvicorn_config
    from meerschaum.connectors.sql._tools import get_sqlalchemy_table

    canonical_host = get_sso_config('canonical_hostname')
    prepend_path = get_sso_config('prepend')
    google_client_id = get_sso_config('google', 'id')
    google_client_secret = get_sso_config('google', 'secret')
    google_callback_slug = get_sso_config('google', 'callback')
    facebook_app_id = get_sso_config('facebook', 'id')
    facebook_app_secret = get_sso_config('facebook', 'secret')
    facebook_callback_slug = get_sso_config('facebook', 'callback')
    db_label = get_sso_config('db_label')
    conn = get_connector('sql', db_label)
    tables = {
        'people': get_sqlalchemy_table('people', connector=conn),
        'people-with-contact-info': get_sqlalchemy_table(
            'people-with-contact-info', connector=conn
        ),
        'session': get_sqlalchemy_table('session', connector=conn),
        'contact_info': get_sqlalchemy_table('contact_info', connector=conn),
    }

    oauth = OAuth()
    oauth.register(
        name='google',
        client_id=google_client_id,
        client_secret=google_client_secret,
        server_metadata_url=GOOGLE_CONF_URL,
        client_kwargs={
            'scope': 'openid email profile',
        },
    )
    oauth.register(
        name='facebook',
        client_id=facebook_app_id,
        client_secret=facebook_app_secret,
        redirect_uri=facebook_callback_slug,
        client_kwargs={
            'scope': 'email',
        },
        **FACEBOOK_CONF_DICT
    )
    app.add_middleware(SessionMiddleware, secret_key=google_client_secret)

    sso_google_path = prepend_path + '/sso/google'
    sso_facebook_path = prepend_path + '/sso/facebook'
    redirect_target = '/'

    def get_address_from_email(email: str) -> Union[str, None]:
        """
        Return a physical address linked to an email address.
        """
        #  q = f"SELECT address FROM \"people-with-contact-info\" WHERE email = '{email}'"
        t = tables['people-with-contact-info']
        q = sqlalchemy.select([t.c.address]).where(t.c.email == email)
        return conn.value(q)

    def get_email_from_name(name: str) -> Union[str, None]:
        """
        Return an email address from a real name.
        """
        fname = name.split(' ')[0]
        lname = ' '.join(name.split(' ')[1:])
        #  q = (
            #  "SELECT first, last, id FROM people "
            #  + f"WHERE first = '{fname}' AND last = '{lname}'"
        #  )
        t = tables['people']
        q = sqlalchemy.select([t.c.first, t.c.last, t.c.id]).where(
            t.c.first == fname and t.c.last = lname
        )
        print(str(q))

        if conn.value(q) is None or True:
            fakeemail = fname + lname + '@<EMAIL>'
            return fakeemail

        return conn.value(q)

    def get_host_from_referer(referer: Optional[str]) -> str:
        """
        Parse the host section of a url from a full url referer.
        """
        if referer is None:
            return canonical_host
        redirect_full_uri = ':'.join(str(referer).split(':')[0:2]) 
        redirect_base_uri = '/'.join(redirect_full_uri.split('/')[0:3])
        return redirect_base_uri

    def record_login_hash(name: str, email: str):
        """
        Record the current login checksum to the database for validation calls later.
        """
        pattern = f'{quote(name)}TO{quote(email)}INTO{canonical_host}'
        print(pattern)
        result = hashlib.md5(pattern.encode()).hexdigest()

        t = tables['session']
        q = sqlalchemy.select([t.c.login_email]).where(t.c.key == result)
        login_exists = conn.value(q) is not None
        if not login_exists:
            q = sqlalchemy.insert(t).values({'key': result, 'login_email': email})
            conn.exec(q)
            print(f'inserted key for user at {email}')

        return result

    @app.get(prepend_path)
    async def testConnection():
        return "<h1>SSO is alive</h1>"

    @app.get(sso_google_path)
    async def google_login(request: Request, referer: Optional[str] = Header(None)):
        redirect_base = get_host_from_referer(referer)
        redirect_uri = redirect_base + google_callback_slug
        if redirect_base.find('staging') != -1:
            print(str(referer) + ' -> ' + redirect_uri)
        return await oauth.google.authorize_redirect(request, redirect_uri)

    @app.get(sso_google_path + '/callback')
    async def google_callback(request: Request):
        token = await oauth.google.authorize_access_token(request)
        user = await oauth.google.parse_id_token(request, token)

        login_token = record_login_hash(user.name, user.email)

        response = RedirectResponse(redirect_target)
        response.set_cookie(key='login_provider', value='google', secure=True)
        response.set_cookie(key='login_name', value=quote(user.name), secure=True)
        response.set_cookie(key='login_email', value=quote(user.email), secure=True)
        response.set_cookie(key='login_id', value=login_token, secure=True)
        response.set_cookie(key='login_name_verified', value=True, secure=True)
        response.set_cookie(key='login_email_verified', value=True, secure=True)
        return response

    @app.get(sso_facebook_path)
    async def facebook_login(request: Request, referer: Optional[str] = Header(None)):
        redirect_base = get_host_from_referer(referer)
        redirect_uri = redirect_base + facebook_callback_slug
        if redirect_base.find('staging') != -1:
            print(str(referer) + ' -> ' + redirect_uri)
        return await oauth.facebook.authorize_redirect(request, redirect_uri)

    @app.get(sso_facebook_path + '/callback')
    async def facebook_callback(request: Request):
        token = await oauth.facebook.authorize_access_token(request)
        access_token = token.get('access_token', None)
        async with httpx.AsyncClient() as client:
            r = await client.get(
                'https://graph.facebook.com/me?fields=id,name,email',
                params={'access_token': access_token}
            )
        name_verified = True
        email_verified = False

        user_dict = r.json()
        name = user_dict.get('name', None)
        ### TODO handle missing name
        email = user_dict.get('email', None)
        if email == None:
            email = get_email_from_name(name) 
        else:
            email_verified = True
        ### TODO handle missing email

        login_token = record_login_hash(name, email if email != None else 'FB-NO-EMAIL')

        response = RedirectResponse(redirect_target)
        response.set_cookie(key='login_provider', value='facebook', secure=True)
        response.set_cookie(key='login_name', value=quote(name), secure=True)
        response.set_cookie(key='login_email', value=quote(email), secure=True)
        response.set_cookie(key='login_id', value=login_token, secure=True)
        response.set_cookie(key='login_name_verified', value=str(name_verified), secure=True)
        response.set_cookie(key='login_email_verified', value=str(email_verified), secure=True)
        return response


#! /usr/bin/env python # -*- coding: utf-8 -*- # vim:fenc=utf-8 Meerschaum plugin to add SSO login functionality to the API. Return a physical address linked to an email address. #  q = f"SELECT address FROM \"people-with-contact-info\" WHERE email = '{email}'" Return an email address from a real name. #  q = ( #  "SELECT first, last, id FROM people " #  + f"WHERE first = '{fname}' AND last = '{lname}'" #  ) Parse the host section of a url from a full url referer. Record the current login checksum to the database for validation calls later. ### TODO handle missing name ### TODO handle missing email