init: initial commit
This commit is contained in:
commit
a1273aa39f
8 changed files with 160 additions and 0 deletions
3
.gitignore
vendored
Normal file
3
.gitignore
vendored
Normal file
|
@ -0,0 +1,3 @@
|
||||||
|
*.env
|
||||||
|
*.bunq_ctx
|
||||||
|
venv/
|
7
README.md
Normal file
7
README.md
Normal file
|
@ -0,0 +1,7 @@
|
||||||
|
# `finmgr`
|
||||||
|
|
||||||
|
Single user finance manager to make sure *someone* doesn't spend too much money.
|
||||||
|
|
||||||
|
## Configure
|
||||||
|
|
||||||
|
check `./doc/env.md` and you'll figure it out!
|
7
doc/env.md
Normal file
7
doc/env.md
Normal file
|
@ -0,0 +1,7 @@
|
||||||
|
| key | description | required | default |
|
||||||
|
|-------------------------|-----------------------------------------------------------|----------|-----------|
|
||||||
|
| `AUTH_USERNAME` | username for basic http auth | no | `admin` |
|
||||||
|
| `AUTH_PASSWORD` | password for basic http auth | no | `admin` |
|
||||||
|
| `BUNQ_API_KEY` | Bunq api key | yes | nothing |
|
||||||
|
| `BUNQ_ENVIRONMENT_TYPE` | Bunq environment type | no | `sandbox` |
|
||||||
|
| `BUNQ_CONTEXT_PREFIX` | Prefix the path of where you want/have your bunq contexts | no | `.` |
|
2
requirements.txt
Normal file
2
requirements.txt
Normal file
|
@ -0,0 +1,2 @@
|
||||||
|
bunq_sdk
|
||||||
|
falcon
|
25
src/AppMeta.py
Normal file
25
src/AppMeta.py
Normal file
|
@ -0,0 +1,25 @@
|
||||||
|
import socket
|
||||||
|
|
||||||
|
APP_NAME = "finmgr"
|
||||||
|
APP_VERSION = "1.0"
|
||||||
|
APP_AUTHORS = [
|
||||||
|
"Raine <raine@ixvd.net>"
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def app_authors_as_string():
|
||||||
|
if len(APP_AUTHORS) == 0:
|
||||||
|
return "No authors"
|
||||||
|
elif len(APP_AUTHORS) == 1:
|
||||||
|
return APP_AUTHORS[0]
|
||||||
|
else:
|
||||||
|
authors = APP_AUTHORS[:-1]
|
||||||
|
return ", ".join(authors) + " and " + APP_AUTHORS[-1]
|
||||||
|
|
||||||
|
|
||||||
|
def app_identifier():
|
||||||
|
return f"{APP_NAME} {APP_VERSION} by {app_authors_as_string()}"
|
||||||
|
|
||||||
|
|
||||||
|
def unique_app_identifier():
|
||||||
|
return f"{app_identifier()} ({socket.gethostname()})"
|
61
src/Bunq.py
Normal file
61
src/Bunq.py
Normal file
|
@ -0,0 +1,61 @@
|
||||||
|
import enum
|
||||||
|
import os
|
||||||
|
from os.path import isfile
|
||||||
|
import socket
|
||||||
|
|
||||||
|
from bunq import ApiEnvironmentType
|
||||||
|
from bunq.sdk.context.api_context import ApiContext
|
||||||
|
from bunq.sdk.context.bunq_context import BunqContext
|
||||||
|
from bunq.sdk.model.generated.endpoint import User, UserLight, UserCompany, UserPerson
|
||||||
|
|
||||||
|
import AppMeta
|
||||||
|
|
||||||
|
|
||||||
|
class Bunq:
|
||||||
|
def __init__(self, environment_type: ApiEnvironmentType, api_key: str):
|
||||||
|
self.environment_type = environment_type
|
||||||
|
self.api_key: str = api_key
|
||||||
|
self.user = None
|
||||||
|
self.setup_context()
|
||||||
|
self.setup_current_user()
|
||||||
|
|
||||||
|
def context_file_path(self):
|
||||||
|
suffix = ""
|
||||||
|
if self.environment_type == ApiEnvironmentType.SANDBOX:
|
||||||
|
suffix = ".sandbox.bunq_ctx"
|
||||||
|
elif self.environment_type == ApiEnvironmentType.PRODUCTION:
|
||||||
|
suffix = ".production.bunq_ctx"
|
||||||
|
return os.path.join(os.environ.get('BUNQ_CONTEXT_PREFIX') or ".", suffix)
|
||||||
|
|
||||||
|
def setup_context(self):
|
||||||
|
if isfile(self.context_file_path()):
|
||||||
|
pass # Config is already present
|
||||||
|
else:
|
||||||
|
ApiContext.create(
|
||||||
|
self.environment_type,
|
||||||
|
self.api_key,
|
||||||
|
AppMeta.unique_app_identifier()
|
||||||
|
).save(self.context_file_path())
|
||||||
|
|
||||||
|
api_context = ApiContext.restore(self.context_file_path())
|
||||||
|
api_context.ensure_session_active()
|
||||||
|
api_context.save(self.context_file_path())
|
||||||
|
|
||||||
|
BunqContext.load_api_context(api_context)
|
||||||
|
|
||||||
|
def setup_current_user(self):
|
||||||
|
user = User.get().value.get_referenced_object()
|
||||||
|
if (isinstance(user, UserPerson)
|
||||||
|
or isinstance(user, UserCompany)
|
||||||
|
or isinstance(user, UserLight)
|
||||||
|
):
|
||||||
|
self.user = user
|
||||||
|
|
||||||
|
def current_user(self):
|
||||||
|
"""
|
||||||
|
:rtype: UserCompany|UserPerson
|
||||||
|
"""
|
||||||
|
return self.user
|
||||||
|
|
||||||
|
def update_context(self):
|
||||||
|
BunqContext.api_context().save(self.context_file_path())
|
31
src/main.py
Normal file
31
src/main.py
Normal file
|
@ -0,0 +1,31 @@
|
||||||
|
import os
|
||||||
|
from wsgiref.simple_server import make_server
|
||||||
|
|
||||||
|
import falcon
|
||||||
|
from dotenv import load_dotenv
|
||||||
|
|
||||||
|
from bunq import ApiEnvironmentType
|
||||||
|
|
||||||
|
from Bunq import Bunq
|
||||||
|
from middleware.AuthMiddleware import AuthMiddleware
|
||||||
|
|
||||||
|
load_dotenv()
|
||||||
|
|
||||||
|
BUNQ = Bunq(
|
||||||
|
ApiEnvironmentType.PRODUCTION if os.environ.get("BUNQ_ENVIRONMENT_TYPE") == "production" else ApiEnvironmentType.SANDBOX,
|
||||||
|
os.environ.get("BUNQ_API_KEY"))
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
print(f"Running for {BUNQ.current_user().display_name}...")
|
||||||
|
|
||||||
|
app = falcon.App()
|
||||||
|
app.add_middleware(AuthMiddleware())
|
||||||
|
|
||||||
|
with make_server('', 8080, app) as server:
|
||||||
|
print(f"Serving on :8080!")
|
||||||
|
server.serve_forever()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
main()
|
24
src/middleware/AuthMiddleware.py
Normal file
24
src/middleware/AuthMiddleware.py
Normal file
|
@ -0,0 +1,24 @@
|
||||||
|
import base64
|
||||||
|
import os
|
||||||
|
|
||||||
|
import falcon
|
||||||
|
from falcon import Request, Response
|
||||||
|
|
||||||
|
|
||||||
|
class AuthMiddleware:
|
||||||
|
def validate_credentials(self, auth_header_value: str) -> bool:
|
||||||
|
auth_token = auth_header_value.split(' ')[1]
|
||||||
|
username, password = base64.b64decode(auth_token).decode().split(':')
|
||||||
|
return username == (os.environ.get('AUTH_USERNAME') or 'admin') and password == (os.environ.get('AUTH_PASSWORD') or 'admin')
|
||||||
|
|
||||||
|
def set_response_to_auth(self, resp: Response):
|
||||||
|
resp.complete = True
|
||||||
|
resp.status = 401
|
||||||
|
resp.body = "Not authenticated."
|
||||||
|
resp.set_header("WWW-Authenticate", "Basic realm='Login required'")
|
||||||
|
|
||||||
|
def process_request(self, request: Request, resp: Response):
|
||||||
|
if request.auth is not None and self.validate_credentials(request.auth):
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
self.set_response_to_auth(resp)
|
Loading…
Reference in a new issue