arrdem/source
1
0
Fork 0
Code Issues 11 Pull requests Projects Releases Packages Wiki Activity

Compare commits

base: arrdem:cb67bdb0515c54552a3c0b5776cceeb9f19fd800
Branches Tags
arrdem:trunk
..
compare: arrdem:d40a20138b3f44c635bd738ba4da377c4ed9c5e8
Branches Tags
arrdem:trunk

No commits in common. "cb67bdb0515c54552a3c0b5776cceeb9f19fd800" and "d40a20138b3f44c635bd738ba4da377c4ed9c5e8" have entirely different histories.
cb67bdb051 ... d40a20138b

7 changed files with 263 additions and 382 deletions
Show stats Expand all files Collapse all files

3
.gitignore vendored
View file

@ -19,8 +19,7 @@
/**/dist /**/dist
/**/node_modules /**/node_modules
bazel-* bazel-*
projects/public-dns/config.yml
public/ public/
tmp/ tmp/
/**/*.sqlite* /**/*.sqlite*
/**/config.toml
/**/config.yml

47
projects/tentacles/src/python/fastmail.py
View file

@ -1,47 +0,0 @@
#!/usr/bin/env python3
from email import encoders
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import smtplib
class FastMailSMTP(smtplib.SMTP_SSL):
"""A wrapper for handling SMTP connections to FastMail."""
def __init__(self, username, password):
super().__init__("mail.messagingengine.com", port=465)
self._username = username
self._password = password
self._has_logged_in = False
def login(self):
if not self._has_logged_in:
super().login(self._username, self._password)
self._has_logged_in = True
def send_message(self, *, from_addr, to_addrs, msg, subject, attachments=None):
self.login()
msg_root = MIMEMultipart()
msg_root["Subject"] = subject
msg_root["From"] = from_addr
msg_root["To"] = ", ".join(to_addrs)
msg_alternative = MIMEMultipart("alternative")
msg_root.attach(msg_alternative)
msg_alternative.attach(MIMEText(msg, "html"))
if attachments:
for attachment in attachments:
prt = MIMEBase("application", "octet-stream")
prt.set_payload(open(attachment, "rb").read())
encoders.encode_base64(prt)
prt.add_header(
"Content-Disposition",
'attachment; filename="%s"' % attachment.replace('"', ""),
)
msg_root.attach(prt)
self.sendmail(from_addr, to_addrs, msg_root.as_string())

10
projects/tentacles/src/python/tentacles/__main__.py
View file

@ -68,10 +68,10 @@ def user_session():
): ):
ctx.sid = session_id ctx.sid = session_id
ctx.uid = uid ctx.uid = uid
user = ctx.db.fetch_user(uid) _id, gid, name, _email, _hash, _status, _verification = ctx.db.fetch_user(uid)
ctx.gid = user.group_id ctx.gid = gid
ctx.username = user.name ctx.username = name
ctx.is_admin = user.group_id == 0 ctx.is_admin = gid == 0
@cli.command() @cli.command()
@ -88,7 +88,7 @@ def serve(hostname: str, port: int, config: Path):
# Before first request # Before first request
create_j2_request_global(app) create_j2_request_global(app)
shutdown_event = create_workers(app, store_factory) shutdown_event = create_workers(lambda: store_factory(app))
# Before request # Before request
app.before_request(user_session) app.before_request(user_session)

42
projects/tentacles/src/python/tentacles/blueprints/user_ui.py
View file

@ -31,17 +31,12 @@ def root():
) )
@BLUEPRINT.route("/user/login", methods=["GET"]) @BLUEPRINT.route("/user/login", methods=["GET", "POST"])
def get_login(): def login():
if is_logged_in(): if is_logged_in():
return redirect("/") return redirect("/")
else: elif request.method == "POST":
return render_template("login.html.j2")
@BLUEPRINT.route("/user/login", methods=["POST"])
def post_login():
if sid := ctx.db.try_login( if sid := ctx.db.try_login(
username := request.form["username"], username := request.form["username"],
salt(request.form["password"]), salt(request.form["password"]),
@ -56,23 +51,21 @@ def post_login():
flash("Incorrect username/password", category="error") flash("Incorrect username/password", category="error")
return render_template("login.html.j2") return render_template("login.html.j2")
else:
return render_template("login.html.j2")
@BLUEPRINT.route("/user/register", methods=["GET"])
def get_register(): @BLUEPRINT.route("/user/register", methods=["GET", "POST"])
def register():
if is_logged_in(): if is_logged_in():
return redirect("/") return redirect("/")
else: elif request.method == "POST":
return render_template("register.html.j2")
@BLUEPRINT.route("/user/register", methods=["POST"])
def post_register():
try: try:
username = request.form["username"] username = request.form["username"]
email = request.form["email"] email = request.form["email"]
group_id = 1 # Normal users group_id = None
status_id = -2 # Unverified status_id = None
for user_config in current_app.config.get("users", []): for user_config in current_app.config.get("users", []):
if user_config["email"] == email: if user_config["email"] == email:
@ -101,6 +94,9 @@ def post_register():
flash("Unable to register that username", category="error") flash("Unable to register that username", category="error")
return render_template("register.html.j2") return render_template("register.html.j2")
else:
return render_template("register.html.j2")
@BLUEPRINT.route("/user/logout") @BLUEPRINT.route("/user/logout")
def logout(): def logout():
@ -111,16 +107,12 @@ def logout():
return resp return resp
@BLUEPRINT.route("/user", methods=["GET"]) @BLUEPRINT.route("/user", methods=["GET", "POST"])
def get_settings(): def settings():
if not is_logged_in(): if not is_logged_in():
return redirect("/") return redirect("/")
return render_template("user.html.j2") elif request.method == "POST":
@BLUEPRINT.route("/user", methods=["POST"])
def post_settings():
if request.form["action"] == "add": if request.form["action"] == "add":
ttl_spec = request.form.get("ttl") ttl_spec = request.form.get("ttl")
if ttl_spec == "forever": if ttl_spec == "forever":

5
projects/tentacles/src/python/tentacles/schema.sql
View file

@ -25,8 +25,7 @@ CREATE TABLE IF NOT EXISTS users (
, email TEXT , email TEXT
, hash TEXT , hash TEXT
, status_id INTEGER , status_id INTEGER
, verified_at TEXT , verification_id INTEGER
, enabled_at TEXT
, FOREIGN KEY(group_id) REFERENCES groups(id) , FOREIGN KEY(group_id) REFERENCES groups(id)
, FOREIGN KEY(status_id) REFERENCES user_statuses(id) , FOREIGN KEY(status_id) REFERENCES user_statuses(id)
, UNIQUE(name) , UNIQUE(name)
@ -36,7 +35,7 @@ CREATE TABLE IF NOT EXISTS users (
---------------------------------------------------------------------------------------------------- ----------------------------------------------------------------------------------------------------
-- Keys represent API keys and auth sessions -- Keys represent API keys and auth sessions
CREATE TABLE IF NOT EXISTS user_keys ( CREATE TABLE IF NOT EXISTS user_keys (
id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(32)))) id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(48))))
, user_id INTEGER , user_id INTEGER
, name TEXT , name TEXT
, expiration TEXT , expiration TEXT

77
projects/tentacles/src/python/tentacles/store.py
View file

@ -111,27 +111,17 @@ class Store(object):
def fetch_user(self, uid: int): def fetch_user(self, uid: int):
return self._conn.execute("SELECT * FROM users WHERE id = ?", [uid]).fetchone() return self._conn.execute("SELECT * FROM users WHERE id = ?", [uid]).fetchone()
@fmap(one)
@requires_conn
def fetch_user_priority(self, uid: int):
return self._conn.execute(
"SELECT priority FROM groups g INNER JOIN users u ON g.id = u.group_id WHERE u.id = ?",
[uid],
).fetchone()
@requires_conn @requires_conn
def list_users(self): def list_users(self):
return self._conn.execute("SELECT * FROM users").fetchall() return self._conn.execute("SELECT id, name FROM users").fetchall()
@requires_conn
def list_unverified_users(self):
return self._conn.execute(
"SELECT * FROM users WHERE status_id = -2 AND NOT verified_at"
).fetchall()
@requires_conn
def verify_user(self, uid: int):
self._conn.execute(
"UPDATE users SET verified_at = datetime('now') WHERE id = ?1", [uid]
)
@requires_conn
def set_user_status(self, uid: int, status):
self._conn.execute(
"UPDATE users SET verified_at = datetime('now') WHERE id = ?1", [uid]
)
@fmap(one) @fmap(one)
@requires_conn @requires_conn
@ -194,13 +184,10 @@ class Store(object):
@requires_conn @requires_conn
def list_keys(self, uid: int): def list_keys(self, uid: int):
return [ return [(id, name, datetime.fromisoformat(exp) if exp else None)
(id, name, datetime.fromisoformat(exp) if exp else None)
for id, name, exp in self._conn.execute( for id, name, exp in self._conn.execute(
"SELECT id, name, expiration FROM user_keys WHERE user_id = ?1 AND name != 'web session'", "SELECT id, name, expiration FROM user_keys WHERE user_id = ?1 AND name != 'web session'", [uid]
[uid], ).fetchall()]
).fetchall()
]
@requires_conn @requires_conn
def fetch_key(self, kid) -> tuple: def fetch_key(self, kid) -> tuple:
@ -292,16 +279,13 @@ class Store(object):
@requires_conn @requires_conn
def update_printer_status(self, printer_id, state: str): def update_printer_status(self, printer_id, state: str):
(status_id,) = self._conn.execute(
"SELECT id FROM printer_statuses WHERE name = ?1", [state]
).fetchone()
self._conn.execute( self._conn.execute(
""" "UPDATE printers SET status_id = ?2, last_poll_date = datetime('now') WHERE id = ?1",
UPDATE printers [printer_id, status_id],
SET
status_id = (SELECT id FROM printer_statuses WHERE name = ?2)
, last_poll_date = datetime('now')
WHERE
id = ?1
""",
[printer_id, state],
) )
################################################################################ ################################################################################
@ -328,9 +312,7 @@ class Store(object):
@requires_conn @requires_conn
def fetch_file(self, uid: int, fid: int): def fetch_file(self, uid: int, fid: int):
return self._conn.execute( return self._conn.execute("SELECT * FROM files WHERE user_id = ?1 AND id = ?2", [uid, fid]).fetchone()
"SELECT * FROM files WHERE user_id = ?1 AND id = ?2", [uid, fid]
).fetchone()
################################################################################ ################################################################################
# Job # Job
@ -350,22 +332,8 @@ class Store(object):
""" """
assert 0 <= relative_priority <= 9 assert 0 <= relative_priority <= 9
return self._conn.execute( return self._conn.execute(
""" "INSERT INTO jobs (user_id, file_id, priority) VALUES (?, ?, ?) RETURNING (id)",
INSERT INTO jobs ( [uid, fid, self.fetch_user_priority(uid) + relative_priority],
user_id
, file_id
, priority
) VALUES (
?1
, ?2
, (
SELECT priority + ?3
FROM users
WHERE uid = ?1
)
) RETURNING (id)
""",
[uid, fid, relative_priority],
).fetchone() ).fetchone()
@requires_conn @requires_conn
@ -501,8 +469,7 @@ class Store(object):
@requires_conn @requires_conn
def cancel_job(self, uid: int, job_id: int): def cancel_job(self, uid: int, job_id: int):
return self._conn.execute( return self._conn.execute(
"UPDATE jobs SET cancelled_at = datetime('now') WHERE user_id = ?1 AND id = ?2", "UPDATE jobs SET cancelled_at = datetime('now') WHERE user_id = ?1 AND id = ?2", [uid, job_id]
[uid, job_id],
) )
@requires_conn @requires_conn

151
projects/tentacles/src/python/tentacles/workers.py
View file

@ -23,10 +23,7 @@ from requests.exceptions import (
HTTPError, HTTPError,
Timeout, Timeout,
) )
from flask import Flask as App, current_app, render_template
from tentacles.store import Store from tentacles.store import Store
from fastmail import FastMailSMTP
class OctoRest(_OR): class OctoRest(_OR):
@ -67,19 +64,15 @@ def corn_job(every: timedelta):
return _decorator return _decorator
def poll_printers(app: App, store: Store) -> None: def poll_printers(db_factory: Callable[[], Store]) -> None:
"""Poll printers for their status.""" """Poll printers for their status."""
for printer in store.list_printers(): with closing(db_factory()) as db:
mapped_job = store.fetch_job_by_printer(printer.id) for printer in db.list_printers():
id, name, url, api_key, last_poll, status = printer
def _set_status(status: str): mapped_job = db.fetch_job_by_printer(id)
if printer.status != status:
print(f"Printer {printer.id} {printer.status} -> {status}")
store.update_printer_status(printer.id, status)
try: try:
client = OctoRest(url=printer.url, apikey=printer.api_key) client = OctoRest(url=url, apikey=api_key)
printer_job = client.job_info() printer_job = client.job_info()
try: try:
printer_state = client.printer().get("state").get("flags", {}) printer_state = client.printer().get("state").get("flags", {})
@ -92,62 +85,70 @@ def poll_printers(app: App, store: Store) -> None:
# polling tasks. This violates separation of concerns a bit, # polling tasks. This violates separation of concerns a bit,
# but appears required for correctness. # but appears required for correctness.
if mapped_job: if mapped_job:
store.finish_job(mapped_job.id, "error") db.finish_job(mapped_job.id, "error")
_set_status("error") if status != "error":
print(f"Printer {printer.id} -> error")
db.update_printer_status(id, "error")
elif printer_state.get("disconnected"): elif printer_state.get("disconnected"):
_set_status("disconnected") if status != "disconnected":
print(f"Printer {printer.id} -> disconnected")
db.update_printer_status(id, "disconnected")
elif printer_state.get("printing"): elif printer_state.get("printing"):
_set_status("running") if status != "running":
print(f"Printer {printer.id} -> running")
db.update_printer_status(id, "running")
elif printer_job.get("state").lower() == "connecting": elif printer_job.get("state").lower() == "connecting":
_set_status("connecting") if status != "connecting":
print(f"Printer {printer.id} -> connecting")
db.update_printer_status(id, "connecting")
elif printer_state.get("ready"): elif printer_state.get("ready"):
_set_status("idle") if status != "idle":
print(f"Printer {printer.id} -> idle")
db.update_printer_status(id, "idle")
else: else:
raise Exception( raise Exception(f"Indeterminate state {printer_job!r} {printer_state!r}")
f"Indeterminate state {printer_job!r} {printer_state!r}"
)
except (ConnectionError, Timeout): except (ConnectionError, Timeout):
_set_status("error") db.update_printer_status(id, "error")
except HTTPError as e: except HTTPError as e:
assert isinstance(e.response, Response) assert isinstance(e.response, Response)
if e.response.status_code in [403, 401] or "error" in printer_job: if e.response.status_code in [403, 401] or "error" in printer_job:
_set_status("error") db.update_printer_status(id, "error")
elif e.response.json().get("error") == "Printer is not operational": elif e.response.json().get("error") == "Printer is not operational":
_set_status("disconnected") db.update_printer_status(id, "disconnected")
else: else:
raise Exception( raise Exception(
f"Indeterminate state {e.response.status_code}, {e.response.json()!r}" f"Indeterminate state {e.response.status_code}, {e.response.json()!r}"
) )
def assign_jobs(app: App, store: Store) -> None: def assign_jobs(db_factory: Callable[[], Store]) -> None:
"""Assign jobs to printers. Uploading files and job state management is handled separately.""" """Assign jobs to printers. Uploading files and job state management is handled separately."""
for printer_id in store.list_idle_printers(): with closing(db_factory()) as db:
if job_id := store.poll_job_queue(): for printer_id in db.list_idle_printers():
store.assign_job(job_id, printer_id) if job_id := db.poll_job_queue():
db.assign_job(job_id, printer_id)
print(f"Mapped job {job_id} to printer {printer_id}") print(f"Mapped job {job_id} to printer {printer_id}")
def push_jobs(app: App, store: Store) -> None: def push_jobs(db_factory: Callable[[], Store]) -> None:
"""Ensure that job files are uploaded and started to the assigned printer.""" """Ensure that job files are uploaded and started to the assigned printer."""
for job in store.list_mapped_jobs(): with closing(db_factory()) as db:
printer = store.fetch_printer(job.printer_id) for job in db.list_mapped_jobs():
file = store.fetch_file(job.user_id, job.file_id) printer = db.fetch_printer(job.printer_id)
file = db.fetch_file(job.user_id, job.file_id)
if not file: if not file:
log.error(f"Job {job.id} no longer maps to a file") log.error(f"Job {job.id} no longer maps to a file")
store.delete_job(job.user_id, job.id) db.delete_job(job.user_id, job.id)
try: try:
client = OctoRest(url=printer.url, apikey=printer.api_key) client = OctoRest(url=printer.url, apikey=printer.api_key)
@ -171,7 +172,7 @@ def push_jobs(app: App, store: Store) -> None:
client.select(Path(file.path).name) client.select(Path(file.path).name)
client.start() client.start()
store.start_job(job.id) db.start_job(job.id)
except TimeoutError: except TimeoutError:
pass pass
@ -179,17 +180,17 @@ def push_jobs(app: App, store: Store) -> None:
log.exception("Oop") log.exception("Oop")
def revoke_jobs(app: App, store: Store) -> None: def revoke_jobs(db_factory: Callable[[], Store]) -> None:
"""Ensure that job files are uploaded and started to the assigned printer. """Ensure that job files are uploaded and started to the assigned printer.
Note that this will ALSO cancel jobs out of the print queue. Note that this will ALSO cancel jobs out of the print queue.
""" """
for job in store.list_cancelled_jobs(): with closing(db_factory()) as db:
for job in db.list_cancelled_jobs():
if job.printer_id: if job.printer_id:
printer = store.fetch_printer(job.printer_id) printer = db.fetch_printer(job.printer_id)
try: try:
print(f"Cancelling running job {job.id}") print(f"Cancelling running job {job.id}")
client = OctoRest(url=printer.url, apikey=printer.api_key) client = OctoRest(url=printer.url, apikey=printer.api_key)
@ -202,8 +203,7 @@ def revoke_jobs(app: App, store: Store) -> None:
raise raise
print(f"Job {job.id} -> cancelled") print(f"Job {job.id} -> cancelled")
store.finish_job(job.id, "cancelled") db.finish_job(job.id, "cancelled")
except TimeoutError: except TimeoutError:
pass pass
@ -212,14 +212,16 @@ def revoke_jobs(app: App, store: Store) -> None:
else: else:
print(f"Unmapped job {job.id} became cancelled") print(f"Unmapped job {job.id} became cancelled")
store.finish_job(job.id, "cancelled") db.finish_job(job.id, "cancelled")
def pull_jobs(app: App, store: Store) -> None:
def pull_jobs(db_factory: Callable[[], Store]) -> None:
"""Poll the state of mapped printers to control jobs.""" """Poll the state of mapped printers to control jobs."""
for job in store.list_running_jobs(): with closing(db_factory()) as db:
printer = store.fetch_printer(job.printer_id) for job in db.list_running_jobs():
printer = db.fetch_printer(job.printer_id)
try: try:
client = OctoRest(url=printer.url, apikey=printer.api_key) client = OctoRest(url=printer.url, apikey=printer.api_key)
job_state = client.job_info() job_state = client.job_info()
@ -233,20 +235,18 @@ def pull_jobs(app: App, store: Store) -> None:
elif job_state.get("progress", {}).get("completion", 0.0) == 100.0: elif job_state.get("progress", {}).get("completion", 0.0) == 100.0:
print(f"Job {job.id} has succeeded") print(f"Job {job.id} has succeeded")
store.finish_job(job.id, "success") db.finish_job(job.id, "success")
elif printer_state.get("error"): elif printer_state.get("error"):
print(f"Job {job.id} has failed") print(f"Job {job.id} has failed")
store.finish_job(job.id, "failed") db.finish_job(job.id, "failed")
elif printer_state.get("cancelling"): elif printer_state.get("cancelling"):
print(f"Job {job.id} has been acknowledged as cancelled") print(f"Job {job.id} has been acknowledged as cancelled")
store.finish_job(job.id, "cancelled") db.finish_job(job.id, "cancelled")
else: else:
print( print(f"Job {job.id} is in a weird state {job_state.get('progress')!r} {printer_state!r}")
f"Job {job.id} is in a weird state {job_state.get('progress')!r} {printer_state!r}"
)
except TimeoutError: except TimeoutError:
pass pass
@ -255,44 +255,15 @@ def pull_jobs(app: App, store: Store) -> None:
log.exception("Oop") log.exception("Oop")
def send_verifications(app, store: Store):
with closing(
FastMailSMTP(
app.config.get("fastmail", {}).get("username"),
app.config.get("fastmail", {}).get("key"),
)
) as fm:
for user in store.list_unverified_users():
fm.send_message(
from_addr="root@tirefireind.us",
to_addrs=[user.email],
msg=render_template("verification_email.html.j2"),
)
def send_approvals(app, store: Store):
with closing(
FastMailSMTP(
app.config.get("fastmail", {}).get("username"),
app.config.get("fastmail", {}).get("key"),
)
):
pass
@corn_job(timedelta(seconds=5)) @corn_job(timedelta(seconds=5))
def run_worker(app, db_factory): def run_worker(db_factory):
with app.app_context(), closing(db_factory(app)) as store: poll_printers(db_factory)
poll_printers(app, store) assign_jobs(db_factory)
assign_jobs(app, store) push_jobs(db_factory)
push_jobs(app, store) revoke_jobs(db_factory)
revoke_jobs(app, store) pull_jobs(db_factory)
pull_jobs(app, store)
send_verifications(app, store)
send_approvals(app, store)
def create_workers(db_factory: Callable[[], Store]) -> Event:
def create_workers(app, db_factory: Callable[[], Store]) -> Event: Thread(target=run_worker, args=[db_factory]).start()
Thread(target=run_worker, args=[app, db_factory]).start()
return SHUTDOWN return SHUTDOWN