mime fix

.env.example

@@ -0,0 +1 @@
+HOST=https://cdn.itguild.info

.gitignore

@@ -2,4 +2,5 @@ packages
 venv
 .idea
 __pycache__
-errlog.txt
+errlog.txt
+.env.local

bearer/models/token.py

@@ -0,0 +1,13 @@
+from typing import Optional, Union
+
+from sqlmodel import Field, SQLModel, Session, select
+from db.DB import DB, DBException
+
+
+class Token(SQLModel):
+    access_token: str
+    token_type: str
+
+
+class TokenData(SQLModel):
+    username: Union[str, None] = None

bearer/routes/auth.py

@@ -0,0 +1,17 @@
+from fastapi import APIRouter, Depends
+from typing import Annotated, Union
+import secrets
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+
+
+auth = APIRouter()
+
+
+@auth.get("/create_secret_key/")
+async def create_secret_key():
+    return {"key": secrets.token_hex(32)}
+
+
+@auth.post("/token")
+async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
+    pass

cdncli.py

@@ -2,7 +2,6 @@ import argparse
 import json
 import requests
 import os
-import magic
 
 parser = argparse.ArgumentParser(description='Videos to images')
 
@@ -70,8 +69,6 @@ if __name__ == '__main__':
 
         filepath = "{dir}/{file}".format(dir=args.dir, file=args.file)
         if os.path.exists(filepath):
-            mime = magic.Magic(mime=True)
-            file_mime = mime.from_file(filepath)
             file = {'file': open(filepath, 'rb')}
             res = requests.post(url='https://cdn.itguild.info/uploadfile/',
                                     files=file, data={'package': d['name'], 'version': d['version'], 'filename': args.file})

daemon/SimpleDaemon.py

@@ -154,6 +154,7 @@ class Daemon:
 
     def console_stdout(self):
         sys.stdout = sys.__stdout__
+        print(sys.stdout.flush())
         print(123)
 
     def process_command(self):

db/DB.py

@@ -0,0 +1,48 @@
+from sqlmodel import Field, SQLModel, create_engine
+from sqlalchemy.ext.asyncio.engine import create_async_engine
+from sqlalchemy.orm.session import sessionmaker
+from sqlalchemy.ext.asyncio.session import AsyncSession
+
+
+class DB:
+    # instance = None
+    config = None
+    engine = None
+    async_engine = None
+    session_maker = None
+
+    def __new__(cls):
+        if not hasattr(cls, 'instance'):
+            cls.instance = super(DB, cls).__new__(cls)
+        return cls.instance
+
+    def __init__(self):
+        from db import config
+        self.config = config
+        sqlite_url = self.config['DATABASE_URI']
+
+        self.engine = create_engine(sqlite_url, echo=True)
+
+        SQLModel.metadata.create_all(self.engine)
+
+    def get_engine(self):
+        return self.engine
+
+    def create_async_engine(self):
+        self.async_engine = create_async_engine(self.config['DATABASE_URI'], furure=True)
+        self.session_maker = sessionmaker(self.async_engine, expire_on_commit=False, class_=AsyncSession)
+
+
+class DBException(Exception):
+    def __init__(self, *args):
+        self.is_db = True
+        if args:
+            self.msg = args[0]
+        else:
+            self.msg = None
+
+    def __str__(self):
+        if self.msg:
+            return "DB error, {0}".format(self.msg)
+        else:
+            return "DB error"

db/__init__.py

@@ -0,0 +1,6 @@
+from dotenv import dotenv_values
+from pathlib import Path  # Python 3.6+ only
+
+env_path = Path('.') / '.env.local'
+
+config = dotenv_values(dotenv_path=env_path)

db/models/User.py

@@ -0,0 +1,52 @@
+from typing import Optional
+
+from sqlmodel import Field, SQLModel, Session, select
+from db.DB import DB, DBException
+from sqlalchemy import exc
+import hashlib, uuid, base64
+
+
+class User(SQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    name: Optional[str] = None
+    login: str = Field(max_length=255)
+    email: str = Field(sa_column_kwargs={'unique': True})
+    age: Optional[int] = None
+    password_hash: str = Field(max_length=255)
+    status: Optional[int] = Field(default=0)
+
+    def create(self):
+        db = DB()
+        session = Session(db.get_engine())
+
+        self.password_hash = self.create_password_hash(self.password_hash)
+        session.add(self)
+
+        try:
+            session.commit()
+            session.refresh(self)
+        except exc.SQLAlchemyError as sqla_error:
+            return DBException(sqla_error)
+
+        session.close()
+
+        return self
+
+    @staticmethod
+    def get_by_id(user_id):
+        db = DB()
+        with Session(db.get_engine()) as s:
+            statement = select(User).where(User.id == user_id)
+            res = s.exec(statement)
+            return res.first()
+
+    @staticmethod
+    def create_password_hash(password):
+        salt = "5gz"
+
+        # Adding salt at the last of the password
+        data_base_password = password + salt
+        # Encoding the password
+        hashed = hashlib.md5(data_base_password.encode())
+
+        return hashed.hexdigest()

main.py

@@ -10,7 +10,7 @@ class CdnServerDaemon(Daemon):
 
 # Press the green button in the gutter to run the script.
 if __name__ == '__main__':
-    # uvicorn.run("server:app", host='0.0.0.0', port=5044, log_level="info")
+    # uvicorn.run("server:app", host='0.0.0.0', port=5044, log_level="info", reload=True)
     with CdnServerDaemon('/tmp/daemon-cdn-server.pid', error_log_file='errlog.txt') as daemon:
         daemon.process_command()
 

mime/mime.py

@@ -0,0 +1,70 @@
+import os
+
+
+def mime_content_type(filename):
+    """Get mime type
+    :param filename: str
+    :type filename: str
+    :rtype: str
+    """
+    mime_types = dict(
+        txt='text/plain',
+        htm='text/html',
+        html='text/html',
+        php='text/html',
+        css='text/css',
+        js='application/javascript',
+        json='application/json',
+        xml='application/xml',
+        swf='application/x-shockwave-flash',
+        flv='video/x-flv',
+
+        # images
+        png='image/png',
+        jpe='image/jpeg',
+        jpeg='image/jpeg',
+        jpg='image/jpeg',
+        gif='image/gif',
+        bmp='image/bmp',
+        ico='image/vnd.microsoft.icon',
+        tiff='image/tiff',
+        tif='image/tiff',
+        svg='image/svg+xml',
+        svgz='image/svg+xml',
+
+        # archives
+        zip='application/zip',
+        rar='application/x-rar-compressed',
+        exe='application/x-msdownload',
+        msi='application/x-msdownload',
+        cab='application/vnd.ms-cab-compressed',
+
+        # audio/video
+        mp3='audio/mpeg',
+        ogg='audio/ogg',
+        qt='video/quicktime',
+        mov='video/quicktime',
+
+        # adobe
+        pdf='application/pdf',
+        psd='image/vnd.adobe.photoshop',
+        ai='application/postscript',
+        eps='application/postscript',
+        ps='application/postscript',
+
+        # ms office
+        doc='application/msword',
+        rtf='application/rtf',
+        xls='application/vnd.ms-excel',
+        ppt='application/vnd.ms-powerpoint',
+
+        # open office
+        odt='application/vnd.oasis.opendocument.text',
+        ods='application/vnd.oasis.opendocument.spreadsheet',
+    )
+
+    ext = os.path.splitext(filename)[1][1:].lower()
+    if ext in mime_types:
+        return mime_types[ext]
+    else:
+        return 'application/octet-stream'

requirements.txt

@@ -20,3 +20,6 @@ uvicorn==0.21.1
 uvloop==0.17.0
 watchfiles==0.18.1
 websockets==10.4
+
+sqlmodel~=0.0.8
+SQLAlchemy~=1.4.41

routes/user_routes.py

@@ -0,0 +1,20 @@
+from fastapi import APIRouter
+from db.models.User import User
+from fastapi import HTTPException
+
+user_route = APIRouter()
+
+
+@user_route.post("/user/", response_model=User, response_model_exclude={"password_hash"})
+async def create_user(user: User):
+    res = user.create()
+    if hasattr(res, 'is_db'):
+        raise HTTPException(status_code=500, detail="DB error {err}".format(err=res.msg))
+
+    return res
+
+
+@user_route.get("/user/{user_id}", response_model=User, response_model_exclude={"password_hash"})
+async def get_user(user_id: int):
+    user = User.get_by_id(user_id)
+    return user

server.py

@@ -1,24 +1,30 @@
 from typing import Union, Annotated
 import uvicorn
+from routes.user_routes import user_route
+from bearer.routes.auth import auth
+from db.DB import DBException
+import mimetypes
+from mime.mime import mime_content_type
 
-from fastapi import FastAPI, File, UploadFile, Form, Response, status
+from fastapi import FastAPI, File, UploadFile, Form, Response, status, HTTPException
 
 import os
 import magic
 
 app = FastAPI()
-
+app.include_router(user_route)
+app.include_router(auth)
 
 @app.get("/")
 def read_root():
     return {"Hello": "World"}
 
 
-@app.get("/items/{version}/{name}/{file}")
+@app.get("/items/{name}/{version}/{file}")
 def read_item(version: str, name: str, file: str = 'index.js'):
-    mime = magic.Magic(mime=True)
     file_path = f"packages/{name}/{version}/{file}"
-    file_mime = mime.from_file(file_path)
+    file_mime = mime_content_type(file_path)
+    print(file_mime)
     if os.path.exists(file_path):
         with open(file_path, 'rb') as f:
             return Response(content=f.read(), media_type=file_mime)
@@ -39,10 +45,18 @@ async def create_upload_file(file: UploadFile | None, package: Annotated[str, Fo
     try:
         contents = file.file.read()
         dir_file = "packages/{package}/{version}".format(package=package, version=version)
+        dir_last = "packages/{package}/latest".format(package=package)
+        dir_stable = "packages/{package}/stable".format(package=package)
         if not os.path.exists(dir_file):
             os.makedirs(dir_file)
+        if not os.path.exists(dir_last):
+            os.makedirs(dir_last)
+        if not os.path.exists(dir_stable):
+            os.makedirs(dir_stable)
         with open(f"{dir_file}/{filename}", 'wb') as f:
             f.write(contents)
+        with open(f"{dir_last}/{filename}", 'wb') as f:
+            f.write(contents)
     except Exception as err:
         return {"message": "There was an error uploading the file, error {err}".format(err=err)}
     finally:

test.py

@@ -0,0 +1,11 @@
+from passlib.hash import bcrypt
+
+
+hashed = bcrypt.using(rounds=13, ident="2y").hash("yoFlOJ5k")
+
+print("$2y$13$C8qbPrEGCbCGgUX9..5BNuzezT3ih9ZFeOgDuYSZ.f7f2SQf5gO0e")
+print(hashed)
+
+res = bcrypt.verify("yoFlOJ5k", "$2y$13$C8qbPrEGCbCGgUX9..5BNuzezT3ih9ZFeOgDuYSZ.f7f2SQf5gO0e")
+
+print(res)