Open yurenchen000 opened 2 years ago
note in demo:
# NOTE
# For demonstration purpose, we use a mock-up globally-shared session object.
change client to use independent sessions, closer to the usual situation.
# 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()
@app.middleware('request')
async def add_session(request):
# request.ctx.session = session
### read session
request.ctx.session = MySessionPool.read_session(request)
# 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
@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.
// 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>
sanic_session
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 cookiesanic-auth
: also at request.ctx.session
$ 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"}}
{}
''' @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) auth.login_user(request, user) 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): auth.logout_user(request) return response.redirect('/login') @app.route('/') @auth.login_required(user_keyword='user1') async def profile(request, user1): content = 'LogoutWelcome, %s
' % user1.name 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) # read user by yourself return response.html(f'you are {user2}') 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) ```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.
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)
{}
''' @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 = 'LogoutWelcome, %s
' % user1.name content += 'Info' 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) ```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
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:
1. test case
I use this examples:
2. operation result
one browser ( chrome ) logged in,
then visit '/', found all other clients (firefox or curl) was loged in.
Is this the expected effect? Am I miss some thing?
3. versions
tested on ubuntu 20.04
with