split out routes
parent
9d7827b28a
commit
04c74bb243
@ -1,238 +1,7 @@
|
||||
import requests
|
||||
import monero.address
|
||||
import monero.seed
|
||||
from quart import Quart, render_template, redirect, request, flash, jsonify
|
||||
from quart_auth import (
|
||||
AuthUser, AuthManager, login_required, login_user, logout_user, current_user, Unauthorized
|
||||
)
|
||||
from quart_bcrypt import Bcrypt
|
||||
from quart_session import Session
|
||||
from lws.factory import create_app
|
||||
|
||||
from lws.models import User, Wallet
|
||||
from lws import config
|
||||
|
||||
app = create_app()
|
||||
|
||||
app = Quart(__name__)
|
||||
app.config["TEMPLATES_AUTO_RELOAD"] = config.TEMPLATES_AUTO_RELOAD
|
||||
app.config["DEBUG"] = config.DEBUG
|
||||
app.config["QUART_ENV"] = config.QUART_ENV
|
||||
app.config["SECRET_KEY"] = config.SECRET_KEY
|
||||
app.config["SESSION_URI"] = config.SESSION_URI
|
||||
app.config["SESSION_TYPE"] = config.SESSION_TYPE
|
||||
app.config["SESSION_PROTECTION"] = config.SESSION_PROTECTION
|
||||
app.config["QUART_AUTH_DURATION"] = config.QUART_AUTH_DURATION
|
||||
Session(app)
|
||||
AuthManager(app)
|
||||
bcrypt = Bcrypt(app)
|
||||
|
||||
|
||||
@app.route("/")
|
||||
@login_required
|
||||
async def index():
|
||||
admin_exists = User.select().first()
|
||||
if not admin_exists:
|
||||
return redirect("/setup")
|
||||
wallets = Wallet.select().order_by(Wallet.date.desc())
|
||||
return await render_template(
|
||||
"index.html",
|
||||
config=config,
|
||||
wallets=wallets
|
||||
)
|
||||
|
||||
|
||||
@app.route("/utils")
|
||||
async def utils():
|
||||
return await render_template("utils/index.html")
|
||||
|
||||
|
||||
@app.route("/utils/mnemonic", methods=["GET", "POST"])
|
||||
async def utils_mnemonic():
|
||||
form = await request.form
|
||||
if form:
|
||||
seed = form.get("seed", "")
|
||||
if not seed:
|
||||
await flash("must provide mnemonic seed")
|
||||
return redirect("/utils/mnemonic")
|
||||
try:
|
||||
s = monero.seed.Seed(seed)
|
||||
return await render_template(
|
||||
"utils/mnemonic.html",
|
||||
results=s
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"failed to read mnemonic seed: {e}")
|
||||
await flash("failed to parse mnemonic seed")
|
||||
return await render_template("utils/mnemonic.html")
|
||||
|
||||
|
||||
@app.route("/login", methods=["GET", "POST"])
|
||||
async def login():
|
||||
if not User.select().first():
|
||||
await flash("must setup first")
|
||||
return redirect("/setup")
|
||||
form = await request.form
|
||||
if form:
|
||||
username = form.get("username", "")
|
||||
password = form.get("password", "")
|
||||
if not username:
|
||||
await flash("must provide a username")
|
||||
return redirect("/login")
|
||||
if not password:
|
||||
await flash("must provide a password")
|
||||
return redirect("/login")
|
||||
user = User.select().where(User.username == username).first()
|
||||
if not user:
|
||||
await flash("this user does not exist")
|
||||
return redirect("/login")
|
||||
pw_matches = bcrypt.check_password_hash(user.password, password)
|
||||
if not pw_matches:
|
||||
await flash("invalid password")
|
||||
return redirect("/login")
|
||||
login_user(AuthUser(user.id))
|
||||
return redirect("/")
|
||||
return await render_template("login.html")
|
||||
|
||||
|
||||
@app.route("/logout")
|
||||
async def logout():
|
||||
if await current_user.is_authenticated:
|
||||
logout_user()
|
||||
return redirect("/")
|
||||
|
||||
@app.route("/setup", methods=["GET", "POST"])
|
||||
async def setup():
|
||||
if User.select().first():
|
||||
await flash("Setup already completed")
|
||||
return redirect("/")
|
||||
form = await request.form
|
||||
if form:
|
||||
username = form.get("username", "")
|
||||
password = form.get("password", "")
|
||||
address = form.get("address", "")
|
||||
view_key = form.get("view_key", "")
|
||||
valid_view_key = False
|
||||
if not username:
|
||||
await flash("must provide a username")
|
||||
return redirect("/setup")
|
||||
if not password:
|
||||
await flash("must provide a password")
|
||||
return redirect("/setup")
|
||||
if not address:
|
||||
await flash("must provide an LWS admin address")
|
||||
return redirect("/setup")
|
||||
if not view_key:
|
||||
await flash("must provide an LWS admin view_key")
|
||||
return redirect("/setup")
|
||||
try:
|
||||
_a = monero.address.Address(address)
|
||||
valid_view_key = _a.check_private_view_key(view_key)
|
||||
except ValueError:
|
||||
await flash("Invalid Monero address")
|
||||
return redirect("/setup")
|
||||
if not valid_view_key:
|
||||
await flash("Invalid view key provided for address")
|
||||
return redirect("/setup")
|
||||
pw_hash = bcrypt.generate_password_hash(password).decode("utf-8")
|
||||
admin = User.create(
|
||||
username=username,
|
||||
password=pw_hash,
|
||||
address=address,
|
||||
view_key=view_key
|
||||
)
|
||||
login_user(AuthUser(admin.id))
|
||||
return redirect("/")
|
||||
return await render_template("setup.html")
|
||||
|
||||
|
||||
@app.route("/wallets")
|
||||
@login_required
|
||||
async def wallets():
|
||||
wallets = Wallet.select().order_by(Wallet.id.asc())
|
||||
return await render_template("wallet/list.html", wallets=wallets)
|
||||
|
||||
@app.route("/wallet/add", methods=["GET", "POST"])
|
||||
@login_required
|
||||
async def wallet_add():
|
||||
form = await request.form
|
||||
if form:
|
||||
name = form.get("name", "")
|
||||
description = form.get("description", "")
|
||||
address = form.get("address", "")
|
||||
view_key = form.get("view_key", "")
|
||||
restore_height = form.get("restore_height", 0)
|
||||
valid_view_key = False
|
||||
if not address:
|
||||
await flash("must provide an LWS admin address")
|
||||
return redirect("/wallet/add")
|
||||
if not view_key:
|
||||
await flash("must provide an LWS admin view_key")
|
||||
return redirect("/wallet/add")
|
||||
try:
|
||||
_a = monero.address.Address(address)
|
||||
valid_view_key = _a.check_private_view_key(view_key)
|
||||
except ValueError:
|
||||
await flash("Invalid Monero address")
|
||||
return redirect("/wallet/add")
|
||||
if not valid_view_key:
|
||||
await flash("Invalid view key provided for address")
|
||||
return redirect("/wallet/add")
|
||||
wallet = Wallet.create(
|
||||
name=name,
|
||||
description=description,
|
||||
address=address,
|
||||
view_key=view_key,
|
||||
restore_height=restore_height,
|
||||
user=User.get(current_user.auth_id)
|
||||
)
|
||||
if not name:
|
||||
wallet.name = f"wallet-{id}"
|
||||
wallet.add_wallet_lws()
|
||||
await flash("wallet added")
|
||||
return redirect(f"/wallet/{wallet.id}")
|
||||
return await render_template("wallet/add.html")
|
||||
|
||||
|
||||
@app.route("/wallet/<id>")
|
||||
@login_required
|
||||
async def wallet_show(id):
|
||||
wallet = Wallet.select().where(Wallet.id == id).first()
|
||||
if not wallet:
|
||||
await flash("wallet does not exist")
|
||||
return redirect("/")
|
||||
return await render_template(
|
||||
"wallet/show.html",
|
||||
wallet=wallet
|
||||
)
|
||||
|
||||
@app.route("/wallet/<id>/rescan")
|
||||
@login_required
|
||||
async def wallet_rescan(id):
|
||||
wallet = Wallet.select().where(Wallet.id == id).first()
|
||||
if not wallet:
|
||||
await flash("wallet does not exist")
|
||||
return redirect("/")
|
||||
wallet.rescan()
|
||||
return redirect(f"/wallet/{id}")
|
||||
|
||||
# / - redirect to /setup if user not setup, to /login if not authenticated
|
||||
# /setup - first time setup user account, encrypted session
|
||||
# /login - log into encrypted session
|
||||
# /wallet/add - add a wallet to LWS
|
||||
# /wallet/:id - show wallet details (balances, txes, etc)
|
||||
# /wallet/:id/remove - remove a wallet from LWS
|
||||
# /wallet/:id/resync - resync wallet
|
||||
|
||||
# get_address_info
|
||||
# get_address_txs
|
||||
# get_random_outs
|
||||
# get_unspent_outs
|
||||
# import_request
|
||||
# submit_raw_tx
|
||||
|
||||
|
||||
@app.errorhandler(Unauthorized)
|
||||
async def redirect_to_login(*_):
|
||||
return redirect("/login")
|
||||
|
||||
def run() -> None:
|
||||
app.run()
|
||||
if __name__ == '__main__':
|
||||
app.run()
|
||||
|
||||
@ -0,0 +1,39 @@
|
||||
from quart import Quart, redirect
|
||||
from quart_auth import (
|
||||
AuthManager, Unauthorized
|
||||
)
|
||||
from quart_bcrypt import Bcrypt
|
||||
from quart_session import Session
|
||||
|
||||
from lws import config
|
||||
|
||||
|
||||
def create_app():
|
||||
app = Quart(__name__)
|
||||
app.config["TEMPLATES_AUTO_RELOAD"] = config.TEMPLATES_AUTO_RELOAD
|
||||
app.config["DEBUG"] = config.DEBUG
|
||||
app.config["QUART_ENV"] = config.QUART_ENV
|
||||
app.config["SECRET_KEY"] = config.SECRET_KEY
|
||||
app.config["SESSION_URI"] = config.SESSION_URI
|
||||
app.config["SESSION_TYPE"] = config.SESSION_TYPE
|
||||
app.config["SESSION_PROTECTION"] = config.SESSION_PROTECTION
|
||||
app.config["QUART_AUTH_DURATION"] = config.QUART_AUTH_DURATION
|
||||
Session(app)
|
||||
AuthManager(app)
|
||||
bcrypt = Bcrypt(app)
|
||||
|
||||
@app.before_serving
|
||||
async def startup():
|
||||
from lws.routes import auth, wallet, meta
|
||||
from lws import filters
|
||||
app.register_blueprint(filters.bp)
|
||||
app.register_blueprint(auth.bp)
|
||||
app.register_blueprint(meta.bp)
|
||||
app.register_blueprint(wallet.bp)
|
||||
|
||||
@app.errorhandler(Unauthorized)
|
||||
async def redirect_to_login(*_):
|
||||
return redirect("/login")
|
||||
return app
|
||||
|
||||
bcrypt = Bcrypt(create_app())
|
||||
@ -0,0 +1,10 @@
|
||||
from monero.numbers import from_atomic
|
||||
from quart import Blueprint
|
||||
|
||||
|
||||
bp = Blueprint('filters', 'filters')
|
||||
|
||||
|
||||
@bp.app_template_filter('atomic')
|
||||
def atomic(amt):
|
||||
return float(from_atomic(amt))
|
||||
@ -0,0 +1,89 @@
|
||||
import monero.address
|
||||
from quart import Blueprint, redirect, request, flash, render_template
|
||||
from quart_auth import login_user, AuthUser, current_user, logout_user
|
||||
|
||||
from lws.factory import bcrypt
|
||||
from lws.models import User
|
||||
|
||||
|
||||
bp = Blueprint('auth', 'auth')
|
||||
|
||||
|
||||
@bp.route("/login", methods=["GET", "POST"])
|
||||
async def login():
|
||||
if not User.select().first():
|
||||
await flash("must setup first")
|
||||
return redirect("/setup")
|
||||
form = await request.form
|
||||
if form:
|
||||
username = form.get("username", "")
|
||||
password = form.get("password", "")
|
||||
if not username:
|
||||
await flash("must provide a username")
|
||||
return redirect("/login")
|
||||
if not password:
|
||||
await flash("must provide a password")
|
||||
return redirect("/login")
|
||||
user = User.select().where(User.username == username).first()
|
||||
if not user:
|
||||
await flash("this user does not exist")
|
||||
return redirect("/login")
|
||||
pw_matches = bcrypt.check_password_hash(user.password, password)
|
||||
if not pw_matches:
|
||||
await flash("invalid password")
|
||||
return redirect("/login")
|
||||
login_user(AuthUser(user.id))
|
||||
return redirect("/")
|
||||
return await render_template("login.html")
|
||||
|
||||
|
||||
@bp.route("/logout")
|
||||
async def logout():
|
||||
if await current_user.is_authenticated:
|
||||
logout_user()
|
||||
return redirect("/")
|
||||
|
||||
|
||||
@bp.route("/setup", methods=["GET", "POST"])
|
||||
async def setup():
|
||||
if User.select().first():
|
||||
await flash("Setup already completed")
|
||||
return redirect("/")
|
||||
form = await request.form
|
||||
if form:
|
||||
username = form.get("username", "")
|
||||
password = form.get("password", "")
|
||||
address = form.get("address", "")
|
||||
view_key = form.get("view_key", "")
|
||||
valid_view_key = False
|
||||
if not username:
|
||||
await flash("must provide a username")
|
||||
return redirect("/setup")
|
||||
if not password:
|
||||
await flash("must provide a password")
|
||||
return redirect("/setup")
|
||||
if not address:
|
||||
await flash("must provide an LWS admin address")
|
||||
return redirect("/setup")
|
||||
if not view_key:
|
||||
await flash("must provide an LWS admin view_key")
|
||||
return redirect("/setup")
|
||||
try:
|
||||
_a = monero.address.Address(address)
|
||||
valid_view_key = _a.check_private_view_key(view_key)
|
||||
except ValueError:
|
||||
await flash("Invalid Monero address")
|
||||
return redirect("/setup")
|
||||
if not valid_view_key:
|
||||
await flash("Invalid view key provided for address")
|
||||
return redirect("/setup")
|
||||
pw_hash = bcrypt.generate_password_hash(password).decode("utf-8")
|
||||
admin = User.create(
|
||||
username=username,
|
||||
password=pw_hash,
|
||||
address=address,
|
||||
view_key=view_key
|
||||
)
|
||||
login_user(AuthUser(admin.id))
|
||||
return redirect("/")
|
||||
return await render_template("setup.html")
|
||||
@ -0,0 +1,49 @@
|
||||
import monero.seed
|
||||
from quart import Blueprint, redirect, request, flash, render_template
|
||||
from quart_auth import login_required
|
||||
|
||||
from lws.factory import bcrypt
|
||||
from lws.models import Wallet, User
|
||||
from lws import config
|
||||
|
||||
|
||||
bp = Blueprint('meta', 'meta')
|
||||
|
||||
@bp.route("/")
|
||||
@login_required
|
||||
async def index():
|
||||
admin_exists = User.select().first()
|
||||
if not admin_exists:
|
||||
return redirect("/setup")
|
||||
wallets = Wallet.select().order_by(Wallet.date.desc())
|
||||
return await render_template(
|
||||
"index.html",
|
||||
config=config,
|
||||
wallets=wallets
|
||||
)
|
||||
|
||||
|
||||
@bp.route("/utils")
|
||||
async def utils():
|
||||
return await render_template("utils/index.html")
|
||||
|
||||
|
||||
@bp.route("/utils/mnemonic", methods=["GET", "POST"])
|
||||
async def utils_mnemonic():
|
||||
form = await request.form
|
||||
if form:
|
||||
seed = form.get("seed", "")
|
||||
if not seed:
|
||||
await flash("must provide mnemonic seed")
|
||||
return redirect("/utils/mnemonic")
|
||||
try:
|
||||
s = monero.seed.Seed(seed)
|
||||
return await render_template(
|
||||
"utils/mnemonic.html",
|
||||
results=s
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"failed to read mnemonic seed: {e}")
|
||||
await flash("failed to parse mnemonic seed")
|
||||
return await render_template("utils/mnemonic.html")
|
||||
|
||||
@ -0,0 +1,94 @@
|
||||
import monero.address
|
||||
from quart import Blueprint, render_template, request, flash, redirect
|
||||
from quart_auth import login_required, current_user
|
||||
|
||||
from lws.models import Wallet
|
||||
|
||||
|
||||
bp = Blueprint('wallet', 'wallet')
|
||||
|
||||
|
||||
@bp.route("/wallets")
|
||||
@login_required
|
||||
async def list():
|
||||
wallets = Wallet.select().order_by(Wallet.id.asc())
|
||||
return await render_template("wallet/list.html", wallets=wallets)
|
||||
|
||||
@bp.route("/wallet/add", methods=["GET", "POST"])
|
||||
@login_required
|
||||
async def add():
|
||||
form = await request.form
|
||||
if form:
|
||||
name = form.get("name", "")
|
||||
description = form.get("description", "")
|
||||
address = form.get("address", "")
|
||||
view_key = form.get("view_key", "")
|
||||
restore_height = form.get("restore_height", 0)
|
||||
valid_view_key = False
|
||||
if not address:
|
||||
await flash("must provide an LWS admin address")
|
||||
return redirect("/wallet/add")
|
||||
if not view_key:
|
||||
await flash("must provide an LWS admin view_key")
|
||||
return redirect("/wallet/add")
|
||||
try:
|
||||
_a = monero.address.Address(address)
|
||||
valid_view_key = _a.check_private_view_key(view_key)
|
||||
except ValueError:
|
||||
await flash("Invalid Monero address")
|
||||
return redirect("/wallet/add")
|
||||
if not valid_view_key:
|
||||
await flash("Invalid view key provided for address")
|
||||
return redirect("/wallet/add")
|
||||
wallet = Wallet.create(
|
||||
name=name,
|
||||
description=description,
|
||||
address=address,
|
||||
view_key=view_key,
|
||||
restore_height=restore_height,
|
||||
user=User.get(current_user.auth_id)
|
||||
)
|
||||
if not name:
|
||||
wallet.name = f"wallet-{id}"
|
||||
wallet.add_wallet_lws()
|
||||
await flash("wallet added")
|
||||
return redirect(f"/wallet/{wallet.id}")
|
||||
return await render_template("wallet/add.html")
|
||||
|
||||
|
||||
@bp.route("/wallet/<id>")
|
||||
@login_required
|
||||
async def show(id):
|
||||
wallet = Wallet.select().where(Wallet.id == id).first()
|
||||
if not wallet:
|
||||
await flash("wallet does not exist")
|
||||
return redirect("/")
|
||||
return await render_template(
|
||||
"wallet/show.html",
|
||||
wallet=wallet
|
||||
)
|
||||
|
||||
@bp.route("/wallet/<id>/rescan")
|
||||
@login_required
|
||||
async def rescan(id):
|
||||
wallet = Wallet.select().where(Wallet.id == id).first()
|
||||
if not wallet:
|
||||
await flash("wallet does not exist")
|
||||
return redirect("/")
|
||||
wallet.rescan()
|
||||
return redirect(f"/wallet/{id}")
|
||||
|
||||
# / - redirect to /setup if user not setup, to /login if not authenticated
|
||||
# /setup - first time setup user account, encrypted session
|
||||
# /login - log into encrypted session
|
||||
# /wallet/add - add a wallet to LWS
|
||||
# /wallet/:id - show wallet details (balances, txes, etc)
|
||||
# /wallet/:id/remove - remove a wallet from LWS
|
||||
# /wallet/:id/resync - resync wallet
|
||||
|
||||
# get_address_info
|
||||
# get_address_txs
|
||||
# get_random_outs
|
||||
# get_unspent_outs
|
||||
# import_request
|
||||
# submit_raw_tx
|
||||
Loading…
Reference in New Issue