sanic-auth

sanic-auth copied to clipboard

1. test case

I use this examples:

• ./examples/note.py // also tried another demo ./examples/blueprint/app.py

2. operation result

one browser ( chrome ) logged in,

then visit '/', found all other clients (firefox or curl) was loged in.

$ curl -sv 'http://localhost:8004'
* Connected to localhost (127.0.0.1) port 8004 (#0)
> GET / HTTP/1.1
> Host: localhost:8004
> User-Agent: curl/7.68.0
> Accept: */*
> 
* Mark bundle as not supporting multiuse
< HTTP/1.1 200 OK
< Content-Length: 48
< Content-Type: text/html; charset=utf-8
< Connection: keep-alive
< Keep-Alive: 5
< 
* Connection #0 to host localhost left intact
<a href="/logout">Logout</a><p>Welcome, demo</p>

Is this the expected effect? Am I miss some thing?

3. versions

tested on ubuntu 20.04

• sanic 20.12.3 + python3.8
• sanic 21.6.2 + python3.9

with

• Sanic-Auth 0.3.0

Jan 21 '22 14:01 yurenchen000

〇. found the answer: it is expected effect.

note in demo:

# NOTE
# For demonstration purpose, we use a mock-up globally-shared session object.

一. some change for demo

change client to use independent sessions, closer to the usual situation.

0. custom SessionPool for test purpose.

# NOTE: just for test purpose, in RAM & no expire
class MySessionPool:
    POOL = {}       # session pool  //in server RAM
    KEY = 'session' # cookie key    //in client cookie

    @classmethod
    def print_pool(c, key=None):
        print('\033[33m -- session_pool:', len(c.POOL))
        for k,v in c.POOL.items():
            print('  -', k, v)
        print('\033[0m')

    @classmethod
    # key from cookie, val from pool[key]
    def read_session(c, request):
        session_key = request.cookies.get(c.KEY, None) # get key from cookie
        if not session_key: 
            session = {}  # dummy
        else:
            session = c.POOL.get(session_key, {})      # get session from pool

        return session

    @classmethod
    # key to cookie, val to pool[key]
    def save_session(c, response, session):
        while (key := secrets.token_urlsafe(32)) in c.POOL: pass

        c.POOL[key] = session         # save val to pool[key]
        response.cookies[c.KEY] = key # save key to client cookie

        c.print_pool()

    @classmethod
    # clr cookie, del session
    def del_session(c, request, response):
        # del session from pool
        key = request.cookies.get(c.KEY, None) 
        if key: c.POOL.pop(key, None)

        # del cookie from client
        response.cookies[c.KEY] = ''
        response.cookies[c.KEY]['max-age'] = 0

        c.print_pool()

1. read session (in middleware) when request start:

@app.middleware('request')
async def add_session(request):
    # request.ctx.session = session
    ### read session
    request.ctx.session = MySessionPool.read_session(request)

2. save session when login success:

# async def login(request):
#     if   ...
            user = User(id=1, name=username)
            auth.login_user(request, user)

            # return response.redirect('/', )
            ### save session
            resp = response.redirect('/')
            MySessionPool.save_session(resp, request.ctx.session)
            return resp

3. clear session when logout:

@app.route('/logout')
@auth.login_required
async def logout(request):
    auth.logout_user(request)
    # return response.redirect('/login')
    ### clr session
    resp = response.redirect('/login')
    MySessionPool.del_session(request, resp)
    return resp

have to call MySessionPool everywhere manually, that's not good. so I wirte demo1_2.py in below reply.

then, every client has their own session.

二. tests

// 0. no login

$ curl -sv 'http://localhost:8004' 
> GET / HTTP/1.1
> Host: localhost:8004
> User-Agent: curl/7.68.0
> Accept: */*
> 

< HTTP/1.1 302 Found
< Location: /login
< content-length: 0
< connection: keep-alive
< content-type: text/html; charset=utf-8
< 
* Connection #0 to host localhost left intact

// 1. login

$ curl -sv 'http://localhost:8004/login' -Fusername=demo -Fpassword=1234
> POST /login HTTP/1.1
> Host: localhost:8004
> User-Agent: curl/7.68.0
> Accept: */*
> Content-Length: 248
> Content-Type: multipart/form-data; boundary=------------------------c5e518168623729b
> 
* We are completely uploaded and fine

< HTTP/1.1 302 Found
< Location: /
< Set-Cookie: session=dGrl3DxPwKJl-AbGuJ04R2bKP4o-UCtG0YJpzpihG7c; Path=/
< content-length: 0
< connection: keep-alive
< content-type: text/html; charset=utf-8
< 
* Connection #0 to host localhost left intact

// 2.1 empty session

$ curl -sv 'http://localhost:8004' 
> GET / HTTP/1.1
> Host: localhost:8004
> User-Agent: curl/7.68.0
> Accept: */*
> 

< HTTP/1.1 302 Found
< Location: /login
< content-length: 0
< connection: keep-alive
< content-type: text/html; charset=utf-8
< 
* Connection #0 to host localhost left intact

// 2.2 loged session

$ curl -sv 'http://localhost:8004'  --cookie session=dGrl3DxPwKJl-AbGuJ04R2bKP4o-UCtG0YJpzpihG7c
> GET / HTTP/1.1
> Host: localhost:8004
> User-Agent: curl/7.68.0
> Accept: */*
> Cookie: session=dGrl3DxPwKJl-AbGuJ04R2bKP4o-UCtG0YJpzpihG7c
> 

< HTTP/1.1 200 OK
< content-length: 48
< connection: keep-alive
< content-type: text/html; charset=utf-8
< 
* Connection #0 to host localhost left intact
<a href="/logout">Logout</a><p>Welcome, demo</p>

Jan 21 '22 19:01 yurenchen000

a simpler & more usable demo with sanic_session

1. code & setup

changes in orig note.py

# NOTE: more practical example, use sanic_session,
#   server side: redis, expire 30 days
#   client side: cookie['session'], httponly

import aioredis
from sanic_session import Session, AIORedisSessionInterface

app.config.redis='redis://:@localhost:6379'
session = Session()

@app.listener('before_server_start')
async def server_init(app, loop):
    app.redis = aioredis.from_url(app.config['redis'], decode_responses=True)
    session.init_app(app, interface=AIORedisSessionInterface(app.redis))


### sanic_session did the job, nothing here
# @app.middleware('request')
# async def add_session(request):
#    request.ctx.session = session

// install depends

$ pip3 install sanic_session[aioredis]
$ sudo apt install redis-server

// redis at localhost:6379, wihout passwd // after sanic restart, sessions still in redis.

nothing else is needed:

• sanic_session will takeover the session manager: server session & client cookie
• and has compatible session interface with sanic-auth: also at request.ctx.session

2. tests

$ curl -sv 'http://localhost:8004/login' -Fusername=demo -Fpassword=1234
> POST /login HTTP/1.1
> Host: localhost:8004
> User-Agent: curl/7.68.0
> Accept: */*
> Content-Length: 248
> Content-Type: multipart/form-data; boundary=------------------------cb999d63dd384851
> 
* We are completely uploaded and fine
* Mark bundle as not supporting multiuse
< HTTP/1.1 302 Found
< Location: /
< Set-Cookie: session=44f888a32a1a4ad0b4b6f4d9f46ea444; Path=/; HttpOnly; expires=Mon, 21-Feb-2022 05:49:23 GMT; Max-Age=2592000
< content-length: 0
< connection: keep-alive
< content-type: text/html; charset=utf-8
< 
* Connection #0 to host localhost left intact

inspect redis:

python3 -c "
import redis
r = redis.Redis()

for key in r.keys('*'):
    typ = r.type(key).decode()
    val = r.get(key).decode() if typ == 'string' else ''
    print(f'--\033[33m {key.decode():52} \033[0m {val}')
"

----
-- session:44f888a32a1a4ad0b4b6f4d9f46ea444              {"TOKEN":{"uid":1,"name":"demo"}}
-- session:jDmBCGT-PSAjoWPQHGufskKyUqv2yruMfUhiAUqU4tI   {"TOKEN":{"uid":1,"name":"demo"}}

Jan 21 '22 22:01 yurenchen000

A shared dict is used intentionally in original example so that there is no external dependency required to run the demo code.

Thank you for your detail explanation and code example, appreciated.

Jan 21 '22 23:01 pyx

thanks for your reply.

I realized that it's designed to flexible and small, without session-manager & authentication-algorithm binding.

I try to write a test demo that simpler (than MySessionPool above & no other depends) and worked (like a normal web login)

// maybe it can be more simper

import secrets
# NOTE: session keeper for test purpose
def session_keeper_demo(app):
    POOL = {}       # session pool  //in server RAM
    KEY = 'session' # cookie key    //in client cookie

    def print_pool():
        print('\033[33m -- session_pool:', len(POOL))
        for k,v in POOL.items(): print('  -', k, v)
        print('\033[0m')

    async def open_session(request):
        '''Before each request: init a session
        '''
        sess_key = request.cookies.get(KEY, None)
        ### 0. open exist session, or new empty
        session = POOL.get(sess_key, {})
        request.ctx.session = session


    async def save_session(request, response):
        '''After each request: save the session, response to client cookies
        '''
        sess_key = request.cookies.get(KEY, None)

        if request.ctx.session:
            if request.ctx.session is not POOL.get(sess_key, {}): 
            ### 1a. new session: save it
                # gen sess_key avoid conflict
                while (new_key := secrets.token_urlsafe(32)) in POOL: pass

                # save session & cookie
                POOL[new_key] = request.ctx.session
                response.cookies[KEY] = new_key
            # else: pass                            
            ### 1b. old session: nothing to do
        else:                                   
            if sess_key: # has cookie: drop it  
            ### 2a. clr session
                # del session & cookie
                POOL.pop(sess_key, None)
                del response.cookies[KEY]  # del cookie (should exist)
            # else: pass 
            ### 2b. nil session: nothing to do

        # print_pool() # log print

    app.request_middleware.appendleft(open_session)
    app.response_middleware.append(save_session)

# session_keeper takeover: session store & cookie handle
session_keeper_demo(app)

from itertools import count
from sanic import Sanic, response
from sanic_auth import Auth, User


app = Sanic(__name__)
app.config.AUTH_LOGIN_ENDPOINT = 'login'
app.config.AUTH_SESSION_NAME = 'SESSION'
app.config.AUTH_TOKEN_NAME = 'TOKEN'
auth = Auth(app)


import secrets
# NOTE: session keeper for test purpose
def session_keeper_demo(app):
    POOL = {}       # session pool  //in server RAM
    KEY = 'session' # cookie key    //in client cookie

    def print_pool():
        print('\033[33m -- session_pool:', len(POOL))
        for k,v in POOL.items(): print('  -', k, v)
        print('\033[0m')

    async def open_session(request):
        '''Before each request: init a session
        '''
        sess_key = request.cookies.get(KEY, None)
        ### 0. open exist session, or new empty
        session = POOL.get(sess_key, {})
        request.ctx.session = session


    async def save_session(request, response):
        '''After each request: save the session, response to client cookies
        '''
        sess_key = request.cookies.get(KEY, None)

        if request.ctx.session:
            if request.ctx.session is not POOL.get(sess_key, {}): 
            ### 1a. new session: save it
                # gen sess_key avoid conflict
                while (new_key := secrets.token_urlsafe(32)) in POOL: pass

                # save session & cookie
                POOL[new_key] = request.ctx.session
                response.cookies[KEY] = new_key
            # else: pass                            
            ### 1b. old session: nothing to do
        else:                                   
            if sess_key: # has cookie: drop it  
            ### 2a. clr session
                # del session & cookie
                POOL.pop(sess_key, None)
                del response.cookies[KEY]  # del cookie (should exist)
            # else: pass 
            ### 2b. nil session: nothing to do

        print_pool() # log print

    app.request_middleware.appendleft(open_session)
    app.response_middleware.append(save_session)

# session_keeper takeover: session store & cookie handle
session_keeper_demo(app)


LOGIN_FORM = '''
<h2>Please sign in, you can try:</h2>
<dl>
<dt>Username</dt> <dd>demo</dd>
<dt>Password</dt> <dd>1234</dd>
</dl>
<p>{}</p>
<form action="" method="POST">
  <input class="username" id="name" name="username"
    placeholder="username" type="text" value=""><br>
  <input class="password" id="password" name="password"
    placeholder="password" type="password" value=""><br>
  <input id="submit" name="submit" type="submit" value="Sign In">
</form>
'''


@app.route('/login', methods=['GET', 'POST'])
async def login(request):
    message = ''
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        # for demonstration purpose only, you should use more robust method
        if username == 'demo' and password == '1234':
            # use User proxy in sanic_auth, this should be some ORM model
            # object in production, the default implementation of
            # auth.login_user expects User.id and User.name available
            user = User(id=1, name=username)
            # print('session0:', session)
            auth.login_user(request, user)
            # print('session1:', session)

            return response.redirect('/', )

        message = 'invalid username or password'
    return response.html(LOGIN_FORM.format(message))


@app.route('/logout')
@auth.login_required
async def logout(request):
    # print('session0:', session)
    auth.logout_user(request)
    # print('session1:', session)
    return response.redirect('/login')

@app.route('/')
@auth.login_required(user_keyword='user1') #  pass current_user_obj to route func as kwarg `user1`
async def profile(request, user1):
    content = '<a href="/logout">Logout</a><p>Welcome, %s</p>' % user1.name
    content += '<a href="/test">Info</a>'
    return response.html(content)
    # return response.html(f'you are {user1}')

@app.route('/test')
@auth.login_required
async def profile2(request):
    user2 = auth.current_user(request)
    return response.html(f'you are {user2}\n')


def handle_no_auth(request):
    return response.json(dict(message='unauthorized'), status=401)


@app.route('/api/user')
@auth.login_required(user_keyword='user', handle_no_auth=handle_no_auth)
async def api_profile(request, user):
    return response.json(dict(id=user.id, name=user.name))


if __name__ == '__main__':
    app.run(host='127.0.0.1', port=8004, debug=True)

Jan 25 '22 22:01 yurenchen000

thanks for your reply.

I realized that it's designed to flexible and small, without session-manager & authentication-algorithm binding.

Yes, the design principle is "separation of concerns".

I wrote a client-side, signed-cookies based session plugin as well, Sanic-CookieSession

Jan 26 '22 01:01 pyx

Yes, the design principle is "separation of concerns".

I wrote a client-side, signed-cookies based session plugin as well, Sanic-CookieSession

glad to see some similar steps in it, which means I'm going the right way.

It's my loss didn't find it at the first time. will try it later.

:+1:

Jan 26 '22 07:01 yurenchen000