diff --git a/.gitignore b/.gitignore index 5d381cc..d353d60 100644 --- a/.gitignore +++ b/.gitignore @@ -160,3 +160,5 @@ cython_debug/ # option (not recommended) you can uncomment the following to ignore the entire idea folder. #.idea/ +config.ini +.idea diff --git a/main.py b/main.py new file mode 100644 index 0000000..5e20ee5 --- /dev/null +++ b/main.py @@ -0,0 +1,108 @@ +import os + +from pathlib import Path +from typing import Annotated +from argon2 import PasswordHasher +from argon2.exceptions import VerifyMismatchError +import configparser + +from fastapi import FastAPI, Depends, HTTPException, status +from fastapi.security import HTTPBasic, HTTPBasicCredentials + +app = FastAPI() + +security = HTTPBasic() + +PATH = '/etc/' +# PATH = './' +USB_DHCPD_FILENAME = 'dhcpcd-usb.conf' +ON_FILENAME = 'dhcpcd-usb-on.conf' +OFF_FILENAME = 'dhcpcd-usb-off.conf' + + +def get_username(credentials: Annotated[HTTPBasicCredentials, Depends(security)]): + config = configparser.ConfigParser() + config.read('./config.ini') + + if credentials.username != config['credentials']['username']: + raise_unauthorized() + + try: + PasswordHasher().verify(config['credentials']['password_hash'], credentials.password) + except VerifyMismatchError: + raise_unauthorized() + + return credentials.username + + +def raise_unauthorized(): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Incorrect username or password", + headers={"WWW-Authenticate": "Basic"}, + ) + + +@app.get('/status') +async def get_status(username: Annotated[str, Depends(get_username)]): + return {"status": "usb" if is_usb_on() else "eth"} + + +@app.post('/status/{status}') +async def set_status(username: Annotated[str, Depends(get_username)], status): + if status == 'usb' and not is_usb_on(): + switch_status(True) + elif status == 'eth' and is_usb_on(): + switch_status(False) + + return {'status': 'usb' if is_usb_on() else 'eth'} + + +def is_usb_on(): + real_filename = os.path.basename(os.path.realpath(PATH + USB_DHCPD_FILENAME)) + return real_filename == ON_FILENAME + + +def switch_status(turn_on): + if turn_on: + # call usb modem api to turn it on + update_usb_modem(turn_on) + + update_dhcp_conf(turn_on) + + if not turn_on: + # call usb modem api to turn it off + update_usb_modem(turn_on) + + +def update_dhcp_conf(turn_on): + os.remove(PATH + USB_DHCPD_FILENAME) + source_file = ON_FILENAME if turn_on else OFF_FILENAME + os.symlink(PATH + source_file, PATH + USB_DHCPD_FILENAME) + + +def update_usb_modem(turn_on): + pass + + +def initialize_app(): + script_dir = Path(__file__).parent.absolute() + credentials_file_path = str(script_dir) + '/config.ini' + if not os.path.isfile(credentials_file_path): + username = input("Enter a username") + password = input("Enter a password") + + config = configparser.ConfigParser() + config['credentials'] = { + 'username': username, + 'password_hash': PasswordHasher().hash(password) + } + with open(credentials_file_path, 'w') as configfile: + config.write(configfile) + + +if __name__ == "__main__": + initialize_app() + + import uvicorn + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..561ed6d --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +fastapi +uvicorn +argon2-cffi