PTV/v6d1tokens
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

reformat

Browse Source
This commit is contained in:
AF 2023-10-02 16:59:39 +00:00
parent bb3ec2e124
commit a3847df309
8 changed files with 48 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
setup.py
View File

@ -1,21 +1,21 @@
from setuptools import setup
setup(
name='v6d1tokens',
version='',
packages=['v6d1tokens'],
url='',
license='',
author='PARRRATE T&V',
author_email='',
description='',
name="v6d1tokens",
version="",
packages=["v6d1tokens"],
url="",
license="",
author="PARRRATE T&V",
author_email="",
description="",
install_requires=[
'aiohttp',
'PyNaCl~=1.4.0',
"aiohttp",
"PyNaCl~=1.4.0",
],
extras_require={
'full': [
'v6d0auth[full] @ git+https://gitea.parrrate.ru/PTV/v6d0auth.git@c718d4d1422945a756213d22d9e26aa24babe0f6',
"full": [
"v6d0auth[full] @ git+https://gitea.parrrate.ru/PTV/v6d0auth.git@c718d4d1422945a756213d22d9e26aa24babe0f6",
],
},
)

18
v6d1tokens/app.py
View File

@ -11,7 +11,7 @@ from v6d0auth.appfactory import *
from v6d0auth.client import *
from v6d1tokens.tdb import *
__all__ = ('V6D1TokensAppFactory',)
__all__ = ("V6D1TokensAppFactory",)
class V6D1TokensAppFactory(AppFactory):
@ -21,9 +21,9 @@ class V6D1TokensAppFactory(AppFactory):
def define_routes(self, routes: web.RouteTableDef):
print(certs.vkey.encode().hex())
@routes.get('/')
@routes.get("/")
async def home(_request: web.Request):
return web.Response(body='v6d1tokens\n')
return web.Response(body="v6d1tokens\n")
async def ws_reg(ws: web.WebSocketResponse):
nonce = random(16)
@ -31,10 +31,10 @@ class V6D1TokensAppFactory(AppFactory):
[token_id, token], hnonce = json.loads(certs.verify(await ws.receive_bytes()))
assert hnonce == nonce.hex()
await self.tdb.reg(token_id, token)
await ws.send_bytes(b'1')
await ws.send_bytes(b"1")
await ws.close()
@routes.get('/reg')
@routes.get("/reg")
async def reg(request: web.Request):
ws = web.WebSocketResponse()
await ws.prepare(request)
@ -42,19 +42,19 @@ class V6D1TokensAppFactory(AppFactory):
return ws
def role_matches_token_id(role: str, token_id: str) -> bool:
return role.startswith(f'token:{token_id}::') or token_id.startswith(f'role:{role}::')
return role.startswith(f"token:{token_id}::") or token_id.startswith(f"role:{role}::")
async def requester_for_request(request: web.Request, requester_cert: bytes, token_id: str) -> VerifyKey:
role = request.headers.get('v6role')
role = request.headers.get("v6role")
if role is None:
return VerifyKey(certs.averify(requester_cert))
else:
requester = VerifyKey(requester_cert)
assert (await has_role(requester, role))
assert await has_role(requester, role)
assert role_matches_token_id(role, token_id)
return requester
@routes.post('/get')
@routes.post("/get")
async def get(request: web.Request):
try:
token_id, requester_hcert = json.loads(await request.read())

25
v6d1tokens/client.py
View File

@ -7,30 +7,27 @@ from v6d0auth import certs
from v6d0auth.client import *
from v6d1tokens.config import taurl
__all__ = ('request_token',)
__all__ = ("request_token",)
async def request_token(query: str, query_mode: Optional[str] = None, tail: str = '') -> str:
async def request_token(query: str, query_mode: Optional[str] = None, tail: str = "") -> str:
async with aiohttp.ClientSession() as session:
if query_mode is None:
if tail:
raise ValueError('tail is used only for non-None query modes')
request = session.post(
f'{taurl}/get',
data=json.dumps([query, (await mycert()).hex()]).encode()
)
raise ValueError("tail is used only for non-None query modes")
request = session.post(f"{taurl}/get", data=json.dumps([query, (await mycert()).hex()]).encode())
else:
if query_mode == 'token':
token_id, role = query, f'token:{query}::{tail}'
elif query_mode == 'role':
token_id, role = f'role:{query}::{tail}', query
if query_mode == "token":
token_id, role = query, f"token:{query}::{tail}"
elif query_mode == "role":
token_id, role = f"role:{query}::{tail}", query
else:
raise ValueError('unknown query mode')
raise ValueError("unknown query mode")
await with_role(role)
request = session.post(
f'{taurl}/get',
f"{taurl}/get",
data=json.dumps([token_id, certs.vkey.encode().hex()]).encode(),
headers={'v6role': role}
headers={"v6role": role},
)
async with request as response:
if response.status == 200:

9
v6d1tokens/config.py
View File

@ -2,8 +2,11 @@ import os
from v6d0auth.config import port, root
__all__ = ('myroot', 'taurl',)
__all__ = (
"myroot",
"taurl",
)
myroot = root / 'v6d1tokens'
myroot = root / "v6d1tokens"
myroot.mkdir(exist_ok=True)
taurl = os.getenv('v6taurl', f'http://127.0.0.1:{port}')
taurl = os.getenv("v6taurl", f"http://127.0.0.1:{port}")

6
v6d1tokens/reg-token.py
View File

@ -9,11 +9,11 @@ from v6d0auth.config import host, port
async def main():
async with aiohttp.ClientSession() as session:
async with session.ws_connect(f'http://{host}:{port}/reg') as ws:
async with session.ws_connect(f"http://{host}:{port}/reg") as ws:
nonce = await ws.receive_bytes()
await ws.send_bytes(certs.sign(json.dumps([[input('token_id:'), input('token:')], nonce.hex()]).encode()))
await ws.send_bytes(certs.sign(json.dumps([[input("token_id:"), input("token:")], nonce.hex()]).encode()))
print((await ws.receive_bytes()).hex())
if __name__ == '__main__':
if __name__ == "__main__":
asyncio.run(main())

2
v6d1tokens/run-server.py
View File

@ -11,7 +11,7 @@ async def main():
await run_app(V6D1TokensAppFactory(tdb).app())
if __name__ == '__main__':
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:

4
v6d1tokens/tdb.py
View File

@ -5,9 +5,9 @@ from nacl.public import PublicKey, SealedBox
from ptvp35 import *
from v6d1tokens.config import myroot
__all__ = ('TDB',)
__all__ = ("TDB",)
_tdbfile = myroot / 'tokens.db'
_tdbfile = myroot / "tokens.db"
class TDB:

8
v6d1tokens/test-request.py
View File

@ -4,10 +4,10 @@ from v6d1tokens.client import *
async def main():
print(await request_token('test'))
print(await request_token('t3st', 'role', '7est'))
print(await request_token('te5t', 'token', 'tes7'))
print(await request_token("test"))
print(await request_token("t3st", "role", "7est"))
print(await request_token("te5t", "token", "tes7"))
if __name__ == '__main__':
if __name__ == "__main__":
asyncio.run(main())