mirror of
https://gitlab.archlinux.org/archlinux/aurweb.git
synced 2025-02-03 10:43:03 +01:00
4c14a10b91
There was one blazing issue with the previous implementation regardless of the multiple records: we were generating fingerprints by storing the key into a file and reading it with ssh-keygen. This is absolutely terrible and was not meant to be left around (it was forgotten, my bad). Took this opportunity to clean up a few things: - simplify pubkey validation - centralize things a bit better Signed-off-by: Kevin Morris <kevr@0cost.org>
113 lines
3.9 KiB
Python
113 lines
3.9 KiB
Python
from typing import Any, Dict
|
|
|
|
from fastapi import Request
|
|
|
|
from aurweb import cookies, db, models, time, util
|
|
from aurweb.models import SSHPubKey
|
|
from aurweb.models.ssh_pub_key import get_fingerprint
|
|
from aurweb.util import strtobool
|
|
|
|
|
|
def simple(U: str = str(), E: str = str(), H: bool = False,
|
|
BE: str = str(), R: str = str(), HP: str = str(),
|
|
I: str = str(), K: str = str(), J: bool = False,
|
|
CN: bool = False, UN: bool = False, ON: bool = False,
|
|
S: bool = False, user: models.User = None,
|
|
**kwargs) -> None:
|
|
now = time.utcnow()
|
|
with db.begin():
|
|
user.Username = U or user.Username
|
|
user.Email = E or user.Email
|
|
user.HideEmail = strtobool(H)
|
|
user.BackupEmail = BE or user.BackupEmail
|
|
user.RealName = R or user.RealName
|
|
user.Homepage = HP or user.Homepage
|
|
user.IRCNick = I or user.IRCNick
|
|
user.PGPKey = K or user.PGPKey
|
|
user.Suspended = strtobool(S)
|
|
user.InactivityTS = now * int(strtobool(J))
|
|
user.CommentNotify = strtobool(CN)
|
|
user.UpdateNotify = strtobool(UN)
|
|
user.OwnershipNotify = strtobool(ON)
|
|
|
|
|
|
def language(L: str = str(),
|
|
request: Request = None,
|
|
user: models.User = None,
|
|
context: Dict[str, Any] = {},
|
|
**kwargs) -> None:
|
|
if L and L != user.LangPreference:
|
|
with db.begin():
|
|
user.LangPreference = L
|
|
context["language"] = L
|
|
|
|
|
|
def timezone(TZ: str = str(),
|
|
request: Request = None,
|
|
user: models.User = None,
|
|
context: Dict[str, Any] = {},
|
|
**kwargs) -> None:
|
|
if TZ and TZ != user.Timezone:
|
|
with db.begin():
|
|
user.Timezone = TZ
|
|
context["language"] = TZ
|
|
|
|
|
|
def ssh_pubkey(PK: str = str(), user: models.User = None, **kwargs) -> None:
|
|
if not PK:
|
|
# If no pubkey is provided, wipe out any pubkeys the user
|
|
# has and return out early.
|
|
with db.begin():
|
|
db.delete_all(user.ssh_pub_keys)
|
|
return
|
|
|
|
# Otherwise, parse ssh keys and their fprints out of PK.
|
|
keys = util.parse_ssh_keys(PK.strip())
|
|
fprints = [get_fingerprint(" ".join(k)) for k in keys]
|
|
|
|
with db.begin():
|
|
# Delete any existing keys we can't find.
|
|
to_remove = user.ssh_pub_keys.filter(
|
|
~SSHPubKey.Fingerprint.in_(fprints))
|
|
db.delete_all(to_remove)
|
|
|
|
# For each key, if it does not yet exist, create it.
|
|
for i, full_key in enumerate(keys):
|
|
prefix, key = full_key
|
|
exists = user.ssh_pub_keys.filter(
|
|
SSHPubKey.Fingerprint == fprints[i]
|
|
).exists()
|
|
if not db.query(exists).scalar():
|
|
# No public key exists, create one.
|
|
db.create(models.SSHPubKey, UserID=user.ID,
|
|
PubKey=" ".join([prefix, key]),
|
|
Fingerprint=fprints[i])
|
|
|
|
|
|
def account_type(T: int = None,
|
|
user: models.User = None,
|
|
**kwargs) -> None:
|
|
if T is not None and (T := int(T)) != user.AccountTypeID:
|
|
with db.begin():
|
|
user.AccountTypeID = T
|
|
|
|
|
|
def password(P: str = str(),
|
|
request: Request = None,
|
|
user: models.User = None,
|
|
context: Dict[str, Any] = {},
|
|
**kwargs) -> None:
|
|
if P and not user.valid_password(P):
|
|
# Remove the fields we consumed for passwords.
|
|
context["P"] = context["C"] = str()
|
|
|
|
# If a password was given and it doesn't match the user's, update it.
|
|
with db.begin():
|
|
user.update_password(P)
|
|
|
|
if user == request.user:
|
|
remember_me = request.cookies.get("AURREMEMBER", False)
|
|
|
|
# If the target user is the request user, login with
|
|
# the updated password to update the Session record.
|
|
user.login(request, P, cookies.timeout(remember_me))
|