jeevanjohnson / local-osu-server Goto Github PK

from mitmproxy import http # type: ignore
from mitmproxy.http import Response

class MyMitmproxy:
    async def request(self, flow: http.HTTPFlow) -> None:
        
        # TODO: Redirect to `localosuserver.com` instead of `*.ppy.sh`
        # Ensures safety of the user's bancho account
        if flow.request.pretty_host.endswith(".ppy.sh"):
            subdomain = flow.request.pretty_host.split(".")[0]
            location = flow.request.url.replace(f"https://{subdomain}.ppy.sh", f"http://localhost:8000/{subdomain}")
            
            flow.response = Response.make(
                status_code=307,
                headers={"Location": location},
            )
            
addons = [MyMitmproxy()]

from fastapi import FastAPI
import uvicorn
import settings
from fastapi.staticfiles import StaticFiles
from contextlib import asynccontextmanager
from fastapi.logger import logger
# import pipreqs

# TODO: Use Relic for data
# The idea is to allow the user to either accept/reject the offer to send data to relic
# so that I have some more data that I can use if their is a bug or something
# labels: enhancement

# TODO: Find an alternative to Jira x Github integration for "TODO's"
# labels: enhancement

@asynccontextmanager
async def lifespan(app: FastAPI):
    if settings.DEVELOPER:
        logger.info("Running on Developer Mode")
    
    # add all necessary routerefs
    import frontend
    import osu_client
    import api
    
    app.include_router(frontend.web_client_router)
    app.include_router(osu_client.bancho_handling_router)
    app.include_router(api.api_authentication_router)

    # static is the default folder that fastapi uses for grabbing
    # javascript and css files
    web_path_for_css_and_javascript_files = '/static'
    app.mount(
        web_path_for_css_and_javascript_files, 
        StaticFiles(directory='./server/frontend/static'), 
        name='static'
    )
    
    # TODO: Add an easier method of connecting without mitmproxy
    # preferably the "cloudflare method"

    if settings.MITMPROXY:
        print("Run the 'mitmproxy.sh'")
    
    yield 
            

app = FastAPI(lifespan=lifespan)

@app.get("/")
def root():
    return {"message": "FastAPI server is running!"}

if __name__ == "__main__":   
    uvicorn.run("main:app", port=8000)#, #reload=True)

from fastapi.routing import APIRouter
from fastapi.responses import JSONResponse
from fastapi import Request, Response
from fastapi import Header, Depends
from uuid import uuid4

bancho_handling_router = APIRouter(default_response_class=Response, prefix="/c")


@bancho_handling_router.post("/")
async def handle_request(request: Request):
    print(request.headers)
    if "osu_token" not in request.headers:
        response = await handle_login(request)

    print("================================")
    return response if response else False
    # else:
    #     TODO response = implement bancho request handler
    # return response


async def handle_login(request: Request):
    login_data = await request.body()
    print(login_data)

    # import hashlib
    # hashlib.md5()

    response = bytearray()

'''
if user not exist -> CREATE new account
'''

    # TODO: Finish Everything Related To Send Packets to the Player

import psutil
from fastapi.responses import JSONResponse
from fastapi.routing import APIRouter

application_router = APIRouter(prefix="/application")

def kill_client() -> None:
    for process in psutil.process_iter():
        if process.name() == "osu!.exe":
            process.kill()

# TODO: the process of killing the process should probably be separated into a different file
@application_router.post("/kill")
async def kill_osu_client():
    try: 
        kill_client()
        return JSONResponse(content={"message": "osu! client killed"}, status_code=200)
    except:
        return JSONResponse(content={"message": "osu! client not found"}, status_code=404)

import psutil
from fastapi.routing import APIRouter
from pathlib import Path
from fastapi import Depends, Response
from fastapi.responses import JSONResponse

api_local_data_fetching_router = APIRouter(prefix="/local_data_fetcher")

# TODO: the process of getting the path should probably be separated into a different file
def get_osu_path() -> Path | None:
    # read every process that is opened and check if the process is osu!
    for process in psutil.process_iter():
        if process.name() == "osu!.exe":
            osu_path = Path(process.cwd())
            
            return osu_path
    
    # if the process is not found, return None
    return None

@api_local_data_fetching_router.get("/osu_path")
async def get_osu_path_route(
    osu_path: Path | None = Depends(get_osu_path)
):
    if osu_path is None:
        return Response(status_code=404, content="osu! path not found")
    
    return JSONResponse(content={"osu_path": str(osu_path)})

from pydantic import BaseModel

class ConfigurationUpdate(BaseModel):
    mitmproxy: bool | None = None
    cloudflare: bool | None = None
    developer: bool | None = None
    osu_path: str | None = None
    display_pp_on_leaderboard: bool | None = None
    rank_scores_by_pp_or_score: bool | None = None
    num_scores_seen_on_leaderboards: int | None = None
    allow_pp_from_modified_maps: bool | None = None
    osu_api_key: str | None = None
    osu_daily_api_key: str | None = None
    osu_api_v2_key: str | None = None
    # TODO: find an alternative to needing the user's osu username and password
    osu_username: str | None = None
    osu_password: str | None = None

from fastapi.routing import APIRouter
from fastapi.routing import APIRouter
from fastapi import Response
from dotenv import set_key
from fastapi import Request, Body, Depends
from .models import configuration as configuration_models

# TODO: find a different place to put this
# because it reality, this should not be in the api section
# in fact this shouldn't be in the server section
# this should be somewhere else in between the web client and the server
# because the web client should be able to access the osu path
# but I haven't figured that out yet

api_configuration_router = APIRouter(prefix="/api/v1/configuration")

def get_configuration(
    from_onboarding: bool = False,
    configuration: dict = Body(...)
) -> configuration_models.ConfigurationUpdate:
    
    if from_onboarding:
        return configuration_models.ConfigurationUpdate(
            osu_path=configuration["osu_path"],
            display_pp_on_leaderboard=configuration["display_pp_on_leaderboard"] == "on",
            rank_scores_by_pp_or_score=configuration["rank_scores_by_pp_or_score"] == "on",
            num_scores_seen_on_leaderboards=int(configuration["num_scores_seen_on_leaderboards"]),
            allow_pp_from_modified_maps=configuration["allow_pp_from_modified_maps"] == "on",
            osu_api_key=configuration["osu_api_key"],
            osu_daily_api_key=configuration["osu_daily_api_key"],
            osu_api_v2_key=configuration["osu_api_v2_key"],
            osu_username=configuration["osu_username"],
            osu_password=configuration["osu_password"],
            mitmproxy=None,
            cloudflare=None,
            developer=None
        )
    else:
        return configuration_models.ConfigurationUpdate(**configuration)

def update_env_file(
    new_configuration: configuration_models.ConfigurationUpdate
) -> None: # needs success case
    
    if new_configuration.osu_path:
        set_key(dotenv_path="./.env", key_to_set="OSU_PATH", value_to_set=new_configuration.osu_path)
    
    if new_configuration.display_pp_on_leaderboard:
        set_key(dotenv_path="./.env", key_to_set="DISPLAY_PP_ON_LEADERBOARD", value_to_set=str(new_configuration.display_pp_on_leaderboard))
    
    if new_configuration.rank_scores_by_pp_or_score:
        set_key(dotenv_path="./.env", key_to_set="RANK_SCORES_BY_PP_OR_SCORE", value_to_set=str(new_configuration.rank_scores_by_pp_or_score))
    
    if new_configuration.num_scores_seen_on_leaderboards:
        set_key(dotenv_path="./.env", key_to_set="NUM_SCORES_SEEN_ON_LEADERBOARDS", value_to_set=str(new_configuration.num_scores_seen_on_leaderboards))
    
    if new_configuration.allow_pp_from_modified_maps:
        set_key(dotenv_path="./.env", key_to_set="ALLOW_PP_FROM_MODIFIED_MAPS", value_to_set=str(new_configuration.allow_pp_from_modified_maps))
    
    if new_configuration.osu_api_key:
        set_key(dotenv_path="./.env", key_to_set="OSU_API_KEY", value_to_set=new_configuration.osu_api_key)
    
    if new_configuration.osu_daily_api_key:
        set_key(dotenv_path="./.env", key_to_set="OSU_DAILY_API_KEY", value_to_set=new_configuration.osu_daily_api_key)
    
    if new_configuration.osu_api_v2_key:
        set_key(dotenv_path="./.env", key_to_set="OSU_API_V2_KEY", value_to_set=new_configuration.osu_api_v2_key)
    
    if new_configuration.osu_username:
        set_key(dotenv_path="./.env", key_to_set="OSU_USERNAME", value_to_set=new_configuration.osu_username)
    
    if new_configuration.osu_password:
        set_key(dotenv_path="./.env", key_to_set="OSU_PASSWORD", value_to_set=new_configuration.osu_password)

    if new_configuration.mitmproxy:
        set_key(dotenv_path="./.env", key_to_set="MITMPROXY", value_to_set=str(new_configuration.mitmproxy))
    
    if new_configuration.cloudflare:
        set_key(dotenv_path="./.env", key_to_set="CLOUDFLARE", value_to_set=str(new_configuration.cloudflare))
        
    if new_configuration.developer:
        set_key(dotenv_path="./.env", key_to_set="DEVELOPER", value_to_set=str(new_configuration.developer))
    

# update .env file
@api_configuration_router.post("/update")
async def update_configuration(
    new_configuration: configuration_models.ConfigurationUpdate = Depends(get_configuration)
):
    update_env_file(new_configuration)
    
    return Response(status_code=200, content="Configuration updated")

import psutil
from fastapi.routing import APIRouter
from pathlib import Path
from fastapi import Depends, Response
from fastapi.responses import JSONResponse

api_local_data_fetching_router = APIRouter(prefix="/local_data_fetcher")

# TODO: the process of getting the path should probably be separated into a different file
def get_osu_path() -> Path | None:
    # read every process that is opened and check if the process is osu!
    for process in psutil.process_iter():
        if process.name() == "osu!.exe":
            osu_path = Path(process.cwd())
            
            return osu_path
    
    # if the process is not found, return None
    return None

@api_local_data_fetching_router.get("/osu_path")
async def get_osu_path_route(
    osu_path: Path | None = Depends(get_osu_path)
):
    if osu_path is None:
        return Response(status_code=404, content="osu! path not found")
    
    return JSONResponse(content={"osu_path": str(osu_path)})

from mitmproxy import http # type: ignore
from mitmproxy.http import Response

DEVSERVER_DOMAIN = "ripple.moe"

class MyMitmproxy:
    async def request(self, flow: http.HTTPFlow) -> None:
        # TODO: use hosts file, to 127.0.0.1 for c4.example.com and etc
        
        # TODO: Redirect to `localosuserver.com` instead of `*.ppy.sh`
        # Ensures safety of the user's bancho account
        if flow.request.pretty_host.endswith(DEVSERVER_DOMAIN):
            subdomain = flow.request.pretty_host.split(".")[0]
            location = flow.request.url.replace(f"https://{subdomain}.{DEVSERVER_DOMAIN}", f"http://localhost:8000/{subdomain}")
            
            flow.response = Response.make(
                status_code=307,

# TODO implement base configuration and aggregator HERE
from sqlmodel import Field, SQLModel, create_engine
import settings

class Accounts(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    username: str = Field(default=None)
    password: str = Field(default=None)
    performance_points: int = Field(default=0)
    country: str | None = Field(default=None)



class Sessions(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None


assert settings.SQLLITE_FILE_NAME.endswith(".db")


sqlite_file_name = f"{settings.SQLLITE_FILE_NAME}.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=True)

SQLModel.metadata.create_all(engine)

from fastapi.routing import APIRouter
from fastapi import Header, Depends

bancho_handling_router = APIRouter(prefix="/c")

def logged_in(osu_token: str | None) -> bool:
    if osu_token:
        return True

    return False


@bancho_handling_router.get('/')
async def handle_request(
    osu_token: str | None = Header(default=None)
):
    
    if not logged_in(osu_token):
        # TODO: Finish Login
        ...
    
    # TODO: Finish Everything Related To Send Packets to the Player

from fastapi.routing import APIRouter
from fastapi import Header, Depends

bancho_handling_router = APIRouter(prefix="/c")

def logged_in(osu_token: str | None) -> bool:
    if osu_token:
        return True

    return False


@bancho_handling_router.get('/')
async def handle_request(
    osu_token: str | None = Header(default=None)
):
    
    if not logged_in(osu_token):
        # TODO: Finish Login
        ...
    
    # TODO: Finish Everything Related To Send Packets to the Player

from fastapi import FastAPI
import uvicorn
import settings
from fastapi.staticfiles import StaticFiles
from contextlib import asynccontextmanager
from fastapi.logger import logger
# import pipreqs

# TODO: Use Relic for data
# The idea is to allow the user to either accept/reject the offer to send data to relic
# so that I have some more data that I can use if their is a bug or something
# labels: enhancement

# TODO: Find an alternative to Jira x Github integration for "TODO's"
# labels: enhancement

@asynccontextmanager
async def lifespan(app: FastAPI):
    if settings.DEVELOPER:
        logger.info("Running on Developer Mode")
    
    # add all necessary routerefs
    import frontend
    import osu_client
    import api
    
    app.include_router(frontend.web_client_router)
    app.include_router(osu_client.bancho_handling_router)
    app.include_router(api.api_authentication_router)

    # static is the default folder that fastapi uses for grabbing
    # javascript and css files
    web_path_for_css_and_javascript_files = '/static'
    app.mount(
        web_path_for_css_and_javascript_files, 
        StaticFiles(directory='./server/frontend/static'), 
        name='static'
    )
    
    # TODO: Add an easier method of connecting without mitmproxy
    # preferably the "cloudflare method"

    if settings.MITMPROXY:
        print("Run the 'mitmproxy.sh'")
    
    yield 
            

app = FastAPI(lifespan=lifespan)

@app.get("/")
def root():
    return {"message": "FastAPI server is running!"}

if __name__ == "__main__":   
    uvicorn.run("main:app", port=8000)#, #reload=True)

import psutil
from fastapi.responses import JSONResponse
from fastapi.routing import APIRouter
import settings
import os
from pathlib import Path
from constants import DEVSERVER_DOMAIN

application_router = APIRouter(prefix="/application")

def kill_client() -> None:
    for process in psutil.process_iter():
        if process.name() == "osu!.exe":
            process.kill()

def launch_client() -> None:
    if not settings.OSU_PATH:
        raise Exception("osu! path not found")
    
    osu_client_path = Path(settings.OSU_PATH) / "osu!.exe"
    
    os.startfile(str(osu_client_path), arguments=f"-devserver {DEVSERVER_DOMAIN}")

# TODO: the process of killing the process should probably be separated into a different file
@application_router.post("/kill")
async def kill_osu_client():
    try: 
        kill_client()
        return JSONResponse(content={"message": "osu! client killed"}, status_code=200)
    except:
        return JSONResponse(content={"message": "osu! client not found"}, status_code=404)
    
@application_router.post("/launch")
async def launch_osu_client():
    #try: 
        launch_client()
        return JSONResponse(content={"message": "osu! client launched"}, status_code=200)
    #except Exception as e:
        #return JSONResponse(content={"message": str(e)}, status_code=404)

from fastapi.routing import APIRouter
from fastapi.templating import Jinja2Templates
from fastapi import Request

templates = Jinja2Templates(directory="./server/frontend/pages")

web_client_router = APIRouter()

@web_client_router.get("/onboarding")
async def first_launch(
    request: Request
):
    # your osu!.exe path
    # do you want pp leaderboards
    # do you want to rank your scores based on pp or score
    # how many scores do you want to see on the leaderboad (max 100)
    # do you want to have the ability to modify maps through the fun orange applciation and gain pp from it
    # osu api key (might can abandon)
    # osu daily api key
    # osu api v2
    
    return templates.TemplateResponse(
        name="onboarding.html", 
        context={"request": request},
    )
    
# TODO: Setup an `open/close osu` button on the home page
# Basically run `osu!.exe -devserver localosuserver.com` in the background

import struct




# TODO implement comprehensive abstraction later
def handle_login(data: bytes) -> bytes:
    struct.

from fastapi import FastAPI
import uvicorn
import settings
from fastapi.staticfiles import StaticFiles
from contextlib import asynccontextmanager
from fastapi.logger import logger
# import pipreqs

# TODO: Use Relic for data
# The idea is to allow the user to either accept/reject the offer to send data to relic
# so that I have some more data that I can use if their is a bug or something
# labels: enhancement

# TODO: Find an alternative to Jira x Github integration for "TODO's"
# labels: enhancement

@asynccontextmanager
async def lifespan(app: FastAPI):
    if settings.DEVELOPER:
        logger.info("Running on Developer Mode")
    
    # add all necessary routerefs
    import frontend
    import osu_client
    import api
    
    app.include_router(frontend.web_client_router)
    app.include_router(osu_client.bancho_handling_router)
    app.include_router(api.api_authentication_router)

    # static is the default folder that fastapi uses for grabbing
    # javascript and css files
    web_path_for_css_and_javascript_files = '/static'
    app.mount(
        web_path_for_css_and_javascript_files, 
        StaticFiles(directory='./server/frontend/static'), 
        name='static'
    )
    
    # TODO: Add an easier method of connecting without mitmproxy
    # preferably the "cloudflare method"

    if settings.MITMPROXY:
        print("Run the 'mitmproxy.sh'")
    
    yield 
            

app = FastAPI(lifespan=lifespan)

@app.get("/")
def root():
    return {"message": "FastAPI server is running!"}

if __name__ == "__main__":   
    uvicorn.run("main:app", port=8000)#, #reload=True)