dev step

main.py

@@ -25,7 +25,7 @@ async def main_page(request: Request, user: User = Depends(User.find_from_cookie
         return RedirectResponse(url='/hold')
     
     tools = [{"name": k, "desc": v}   for k, v in chat_functions.function_user_text.items()]
-    token = user.create_token()
+    token = user.create_shortlived_token()
     response = templates.TemplateResponse(
         "main.html", {
             "request": request, 
@@ -33,7 +33,6 @@ async def main_page(request: Request, user: User = Depends(User.find_from_cookie
             "version": fecha_unix,
             "tools": tools
         })
-    user.update()
     return response
 
 
@@ -49,18 +48,18 @@ async def validate_oauth(request: Request):
     params = dict(request.query_params)
     fingerprint = params["state"].split("=",1)[1]
     google_userinfo = oauth.validate_redirect(params) | {"fprint": fingerprint}
-    user = model.User.find_or_create(google_userinfo)
-    token = user.create_token()
+    user:User = model.User.find_or_create(google_userinfo)
+    token = user.create_cookie()
     response = RedirectResponse(url='/')
     response.set_cookie(key="token", value=token)
-    user.update()
+    user.update_user()
     return response
 
 
 ########################## API ##########################    
 
 @app.post("/getConfigs")
-async def get_configs(request: Request, user = Depends(model.User.find_from_header)):
+async def get_configs(request: Request, user ): #= Depends(model.User.find_from_header)):
     tokens = user.tokens
     year, month = datetime.now().strftime("%y-%m").split("-")
     mes = sum(tokens.get(year, {}).get(month, {"_":0}).values())
@@ -68,27 +67,24 @@ async def get_configs(request: Request, user = Depends(model.User.find_from_head
     return user.configs.model_dump() | {"tokens": tokens}
 
 @app.post("/setConfigs")
-async def set_configs(request: Request, configs: model.Configs, user = Depends(User.find_from_header)):
+async def set_configs(request: Request, configs: model.Configs, user ): # = Depends(User.find_from_header)):
     user.configs = configs
     user.update()
     return {"success":True}
     
 
 @app.post("/getToken")
-async def get_token(request: Request, data: dict, user: User = Depends(User.find_from_header)):
-    user.session = model.Session(**data)
-    data.pop("fingerprint")
-    security.validate_signature(**data)
-    security.create_jwt_token(data)
-    token = user.create_token()
-class Session(BaseModel):
-    gid: str
-    fingerprint: str
-    public_key: str
-    return {"su":True}
+async def get_token(request: Request, data: model.APISession = Depends(model.APISession.validate)):
+    # user.session = model.Session(**data)
+    # data.pop("fingerprint")
+    # security.validate_signature(**data)
+    # security.create_jwt_token(data)
+    # token = user.create_token()
+    pass
+
 
 @app.post("/chat")
-async def chat_async(request: Request, body: model.Chat, user = Depends(model.User.find_from_header)):
+async def chat_async(request: Request, body: model.Chat, user ): # = Depends(model.User.find_from_header)):
     if(len(body.messages) < 1 or body.messages[-1].content==""):
         log_module.logger.warning(user.gid, "Mensaje RQ", "error, mensajes vacios")
         raise HTTPException(
@@ -106,20 +102,26 @@ async def chat_async(request: Request, body: model.Chat, user = Depends(model.Us
 async def privacy_policy(request: Request):
     return templates.TemplateResponse("PrivacyPolicy.html", {"request": request})
 
-@app.route("/hold", ["GET", "POST"])
-async def on_hold(request: Request):
+@app.get("/hold") #, ["GET", "POST"]
+async def on_hold(request: Request, user: User = Depends(User.find_from_cookie)):
     redirect_to_login = RedirectResponse(url='/login')
     
-    user = model.User.find_from_token(request, "view")
-
     if user.can_use("chat"):
         return redirect_to_login
-    
-    if request.method=="POST" and not user.description:
+      
+    return templates.TemplateResponse(
+        "no_access.html", {
+            "request": request,
+            "description": bool(user.description)
+        })
+
+@app.post("/hold") #, ["GET", "POST"]
+async def on_hold(request: Request, user: User = Depends(User.find_from_cookie)):
+    if not user.description:
         form = await request.form() 
         if message := form["message"].strip():
             user.update_description(message)
-    
+
     return templates.TemplateResponse(
         "no_access.html", {
             "request": request,

modules/model.py

@@ -1,4 +1,6 @@
-import os 
+import os
+from typing_extensions import Unpack
+from pydantic.config import ConfigDict 
 from pymongo.mongo_client import MongoClient
 from pymongo.server_api import ServerApi
 from fastapi import Request
@@ -60,47 +62,44 @@ class Chat(BaseModel):
     def new_msg(self: Self):
         return Message(role="", content="")
 
-class Session(BaseModel):
+class APISession(BaseModel):
     gid: str
-    fingerprint: str
+    fprint: str
     public_key: str
     guid: str = str(uuid.uuid4())
-    created: datetime = datetime.now(tz)
-    updated: datetime = datetime.now(tz)
-    expire: datetime =  datetime.now(tz)+timedelta(days=7)
-    
-
-
-    @classmethod
-    def remove_expired(cls: Self, gid: str) -> None:
-        DB.sess.delete_one({"gid": gid, "expire": {"$lt": datetime.now(tz)}})
+    challenge: str = str(uuid.uuid4())
 
-    @classmethod
-    def clean_other_sessions(cls:Self, gid: str) -> None:
-        count = DB.sess.count_documents({"gid": gid})
-        if count > 3:
-            ids_a_borrar = [
-                registro["_id"] for registro in DB.sess.find({"gid": gid}).sort({"expire": -1}).skip(3)
-            ]
-            DB.sess.delete_many({"_id": {"$in": ids_a_borrar}})
+    def __init__(self, **kwargs: dict):
+        kwargs["fprint"] = security.sha256(kwargs["fprint"])
+       
+        super(APISession, self).__init__(**kwargs)
+        
+    
+  
 
     @classmethod
-    def find(cls: Self, gid: str, guid: str) -> Self | None:
-        cls.remove_expired(gid)
-        sessj: dict = DB.sess.find_one({"gid": gid, "guid": guid})
-        if sessj:
-            sess: Self = cls(**sessj)
-        sess.updated = datetime.now(tz)
-        sess.expire = datetime.now(tz)+timedelta(days=7)
-        cls.clean_other_sessions(gid)
-        DB.sess.update_one({"gid": gid, "guid": guid}, {"$set": sess.model_dump()})
-        return sess
+    def validate(cls: Self, request: Request, data: dict):
+        token_data:dict = security.token_from_headers(request)
+        if token_data["step"] != "intermediate":
+            security.raise_401()
+        token = request.headers.get("Autorization", " ").split(" ",1)[1]
+        fprint = security.sha256(data["fingerprint"])
+        security.validate_signature(data["public_key"],data["signature"], token)
+        if fprint != token_data["fprint"]:
+            security.raise_401()
+        
+        new_obj = cls(**data)
+        return new_obj 
     
-    @classmethod
-    def create(cls: Self, **kwargs: dict) -> Self:
-        temp:Self = cls(**kwargs)
-        DB.sess.insert_one(temp.model_dump())    
-        return temp
+    def validate_signature(self: Self, signature:str):
+        valid = security.validate_signature(self.public_key, signature, self.challenge)
+        if not valid:
+            security.raise_401()
+        return True
+    
+    def create_token(self: Self):
+        self.challenge = str(uuid.uuid4()) 
+        return security.create_jwt_token(self.model_dump())
 
     
 class User(BaseModel):
@@ -112,8 +111,8 @@ class User(BaseModel):
     email: str
     gid: str
     role: str = "on hold"
+    fprint: str = ""
     configs: Configs = Configs()
-    session: Session | None = None
 
     @classmethod
     def find(cls: Self, gid: str) -> Self | None:
@@ -124,56 +123,35 @@ class User(BaseModel):
     
     @classmethod
     def find_or_create(cls: Self, data: dict)-> Self:
-        sess:dict = Session.create(gid=data["gid"], fprint=data.pop("fprint")) 
+        if not data.get("fprint", None):
+            security.raise_307()
+        data["fprint"] = security.sha256(data["fprint"])
         found:Self = cls.find(data["gid"])
         if found:
-            found.session = sess
+            found.fprint = data["fprint"]
             return found
         user:Self = cls(**data)
-        user.session = sess
-        
-        DB.user.insert_one(user.model_dump(exclude={"session"}))  
+        DB.user.insert_one(user.model_dump(exclude={"session", "fprint"}))  
         log_module.logger.info(f"User created: {user.gid}")
         return user
     
-
-    @classmethod
-    def find_from_header(cls:Self, request: Request) -> Self:
-        temp = cls.find_from_token(request, "api")
-        return temp
-
+    def create_cookie(self: Self) -> str:
+        return security.create_jwt_token({"gid": self.gid, "fprint":self.fprint })
+    
+    def create_shortlived_token(self: Self) -> str:
+        return security.create_jwt_token({"gid": self.gid, "fprint":self.fprint, "step":"intermediate" }, maxlife=3)
+    
+    
+  
     @classmethod
     def find_from_cookie(cls:Self, request: Request) -> Self:
-        return cls.find_from_token(request, "view")
+        data:dict = security.token_from_cookie(request)
+        user:dict = DB.user.find_one({"gid":data["gid"]})
 
-    @classmethod
-    def find_from_token(cls, request: Request, usage: str) -> Self:
-        if usage=="view":
-            data:dict = security.token_from_cookie(request)
-        else:
-            data:dict = security.token_from_headers(request)
-        
-        userj:dict = DB.user.find_one({"gid":data["gid"]})
-
-        if gid := userj.get("gid", None):
-            sess:Session = Session.find(gid, data["guid"])
-        
-        if not gid or not sess:
-            security.raise_307()
-
-        user:Self = cls(**userj)
-        user.session = sess
-        return user
-        
-
-    def create_token(self: Self) -> str:
-        return security.create_jwt_token({
-            "gid": self.gid, 
-            "guid": self.session.guid,
-            "fp": self.session.fprint,
-            "pubk": self.session.public_key
-            })
-        
+        if (user.get("gid", None)):
+            return cls(**user)
+            
+        security.raise_307()
     
     def update_description(self: Self, message: str) -> None:
         log_module.logger.info(f"Description Updated: {self.gid}")
@@ -186,25 +164,11 @@ class User(BaseModel):
     def can_use(self: Self, activity: str):
         return security.can_use(self.role, activity)
     
-    def update(self: Self, message: Message = None) -> None:
-        count_tokens:int = 0
-        if message:
-            count_tokens:int = message.tokensOutput+message.tokensPrompt
-        year, month, day = datetime.now().strftime("%y-%m-%d").split("-")
-        
-        if year not in self.tokens:
-            self.tokens[year] = {}
-
-        if month not in self.tokens[year]:
-            self.tokens[year][month] = {}
-        
-        if day not in self.tokens[year][month] :
-            self.tokens[year][month][day] = 0
-            
-        self.tokens[year][month][day] += count_tokens
-
+    def update_user(self: Self, message: Message = None) -> None:
         DB.user.update_one({"gid": self.gid}, {"$set": self.model_dump(exclude={"guid", "session"})})
 
-
-    
-
+    @staticmethod
+    def update_usage(gid:str, message:Message):
+        count_tokens:int = message.tokensOutput+message.tokensPrompt
+        inc_field = datetime.now().strftime("tokens.%y.%m.%d")
+        DB.user.update_one({"gid": gid}, {"$inc":{inc_field: count_tokens}})

modules/security.py

@@ -19,10 +19,9 @@ for key in settings.USERS:
 
 
 
-def create_jwt_token(data):
-    to_encode = data
-    expire = datetime.utcnow() + timedelta(minutes=settings.JWT_EXPIRATION_TIME_MINUTES_API)    
-    to_encode.update({"exp": expire})
+def create_jwt_token(data, maxlife=settings.JWT_EXPIRATION_TIME_MINUTES_API):
+    expire = datetime.utcnow() + timedelta(minutes=maxlife)    
+    to_encode = data | {"exp": expire}
     encoded_jwt = jwt.encode(to_encode, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)
     return encoded_jwt
 
@@ -79,5 +78,10 @@ def validate_signature(public_key: str, signature: str, data:str) -> bool:
         pss.new(public_key).verify(data_, signature)
         return True
     except ValueError:
-        return False
-    
+        raise_401()
+    
+def sha256(data:str|bytes) -> str:
+    data_ = data if isinstance(data, bytes) else data.encode()
+    return SHA256.new(data_).hexdigest()
+
+

modules/settings.py

@@ -22,7 +22,7 @@ USERS = json.loads(str(os.getenv("USER_KEYS")).replace("\n", ""))
 
 JWT_SECRET = USERS["master"]
 JWT_ALGORITHM = "HS256"
-JWT_EXPIRATION_TIME_MINUTES_API = 30
+JWT_EXPIRATION_TIME_MINUTES_API = 7*24*60
 JWT_EXPIRATION_TIME_MINUTES_VIEW = 7*24*60
 
 

static/js/chatHandler.js

@@ -53,8 +53,7 @@ class ChatGPT{
         let data = {
             fingerprint: this.fp,
             public_key: this.pubk,
-            signature: this.tokenSigned,
-            data: this.token
+            signature: this.tokenSigned
         }
 
         $.ajax({

templates/main.html

@@ -173,7 +173,7 @@
         </dialog>
 
         <script> 
-        async function generatex(){
+        async function start(){
             let keypair = await window.crypto.subtle.generateKey(
                 {
                     name: "RSA-PSS",
@@ -202,7 +202,7 @@
 
             let exportKey_u8 = new Uint8Array(exportKey_ba);
             let exportKey = "-----BEGIN PUBLIC KEY-----\n"+btoa(String.fromCharCode(...exportKey_u8))+"\n-----END PUBLIC KEY-----"
-
+ 
             console.log("exportedKey");
             console.log(exportKey);
             console.log("sign");
@@ -217,7 +217,7 @@
             cHand = new ChatGPT("{{ token }}", fp.visitorId, exportKey, sign);
 
         }
-        generatex()
+        start()