Jonathan Kirkwood
ae967494bd
The external fund_administrator role (relabeled Administrator) now signs
into the full admin interface, fenced to the funds and SPVs granted to
it via EntityAccess:
- Partners, capital accounts, documents (upload and delete), entity
edits, and eNAV imports for its own funds only; no fund creation,
valuation sign-off, audit log, or investor view.
- Scoped user management: sees and manages only investors tied to its
funds; creates investor accounts only; updates preserve grants on
funds outside its scope.
- New DELETE /api/users/{id} (in-app Delete user button) with the
cascade cleanup factored out of the CLI; Service Admin and self are
protected, and an Administrator can only delete an investor who
belongs solely to its funds.
- Internal fund_admin relabeled 'Staff (all funds)' and dropped from
the create picker to end the two-similar-names confusion.
- Version badge removed from the UI (sidebar and portal header); the
build version now logs to the browser console instead.
- deploy/.startos (signing key) added to .gitignore.
368 lines
15 KiB
Python
368 lines
15 KiB
Python
"""CLI commands for Ten31Portal (also driven by the StartOS service Actions)."""
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
|
|
from sqlmodel import Session, select
|
|
|
|
from ten31portal import config
|
|
from ten31portal.auth import hash_password
|
|
from ten31portal.database import engine
|
|
from ten31portal.db_init import run_migrations
|
|
from ten31portal.models import User, UserRole
|
|
|
|
|
|
def _find_user(session: Session, username: str | None, email: str | None) -> User | None:
|
|
"""Look a user up by username (preferred) or email."""
|
|
if username:
|
|
u = session.exec(select(User).where(User.username == username)).first()
|
|
if u:
|
|
return u
|
|
if email:
|
|
return session.exec(select(User).where(User.email == email)).first()
|
|
return None
|
|
|
|
|
|
def create_user(args: argparse.Namespace) -> None:
|
|
"""Create a new user account."""
|
|
run_migrations()
|
|
|
|
with Session(engine) as session:
|
|
if session.exec(select(User).where(User.username == args.username)).first():
|
|
print(f"Error: user with username {args.username} already exists.", file=sys.stderr)
|
|
sys.exit(1)
|
|
if args.email and session.exec(select(User).where(User.email == args.email)).first():
|
|
print(f"Error: user with email {args.email} already exists.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
try:
|
|
role = UserRole(args.role)
|
|
except ValueError:
|
|
print(f"Error: invalid role '{args.role}'. Must be one of: {', '.join(r.value for r in UserRole)}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
user = User(
|
|
name=args.name,
|
|
username=args.username,
|
|
email=args.email or None,
|
|
password_hash=hash_password(args.password),
|
|
role=role,
|
|
is_service_admin=bool(getattr(args, "service_admin", False)),
|
|
)
|
|
session.add(user)
|
|
session.commit()
|
|
print(f"Created user: {user.name} ({user.username}) with role {user.role.value}")
|
|
|
|
|
|
def _clear_admin_password_file() -> None:
|
|
"""Remove the recorded initial admin password (it is no longer valid once changed)."""
|
|
try:
|
|
os.remove(config.ADMIN_PASSWORD_FILE)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def reset_password(args: argparse.Namespace) -> None:
|
|
"""Reset a user's password (and re-enable their login)."""
|
|
run_migrations()
|
|
|
|
with Session(engine) as session:
|
|
user = _find_user(session, args.username, args.email)
|
|
if user is None:
|
|
who = args.username or args.email
|
|
print(f"Error: no user found for '{who}'.", file=sys.stderr)
|
|
sys.exit(1)
|
|
user.password_hash = hash_password(args.password)
|
|
user.login_enabled = True
|
|
user.must_change_password = False
|
|
is_admin = user.is_service_admin
|
|
session.add(user)
|
|
session.commit()
|
|
# Resetting the built-in admin invalidates the initial password recorded on first boot.
|
|
if is_admin:
|
|
_clear_admin_password_file()
|
|
print(f"Password reset for {user.name} ({user.username}).")
|
|
|
|
|
|
def reset_2fa(args: argparse.Namespace) -> None:
|
|
"""Clear a user's two-factor enrollment so they can sign in with password alone."""
|
|
run_migrations()
|
|
|
|
with Session(engine) as session:
|
|
user = _find_user(session, args.username, args.email)
|
|
if user is None:
|
|
who = args.username or args.email
|
|
print(f"Error: no user found for '{who}'.", file=sys.stderr)
|
|
sys.exit(1)
|
|
if not user.totp_enabled and not user.totp_secret:
|
|
print(f"{user.name} ({user.username}) does not have two-factor enabled.")
|
|
return
|
|
user.totp_secret = None
|
|
user.totp_enabled = False
|
|
user.totp_recovery_codes = None
|
|
session.add(user)
|
|
session.commit()
|
|
print(f"Two-factor disabled for {user.name} ({user.username}). "
|
|
"They can sign in with just their password and re-enroll from the app.")
|
|
|
|
|
|
def enable_investor_logins(args: argparse.Namespace) -> None:
|
|
"""Give every no-login investor account the shared default password and enable sign-in.
|
|
|
|
Targets investor-role accounts with login_enabled=False that log in on their own
|
|
(linked secondary names are skipped — they sign in under their primary). Accounts that
|
|
already have a working login are never touched.
|
|
"""
|
|
run_migrations()
|
|
|
|
from ten31portal.models import UserRole
|
|
|
|
with Session(engine) as session:
|
|
users = session.exec(
|
|
select(User).where(
|
|
User.role == UserRole.investor,
|
|
User.login_enabled == False, # noqa: E712 — SQL expression
|
|
User.primary_account_id == None, # noqa: E711
|
|
).order_by(User.name) # type: ignore[arg-type]
|
|
).all()
|
|
if not users:
|
|
print("Nothing to do — every investor account already has a login.")
|
|
return
|
|
for u in users:
|
|
u.password_hash = hash_password(config.DEFAULT_INVESTOR_PASSWORD)
|
|
u.login_enabled = True
|
|
u.must_change_password = True
|
|
session.add(u)
|
|
session.commit()
|
|
for u in users:
|
|
print(f"Enabled login for {u.name} ({u.username})")
|
|
print(
|
|
f"\n{len(users)} investor account(s) set to the default password "
|
|
f"'{config.DEFAULT_INVESTOR_PASSWORD}'. Each investor should change it in the "
|
|
"portal (Change password)."
|
|
)
|
|
|
|
|
|
def show_admin_password(args: argparse.Namespace) -> None:
|
|
"""Print the randomly-generated initial admin password recorded on first boot."""
|
|
path = config.ADMIN_PASSWORD_FILE
|
|
if not os.path.exists(path):
|
|
print(
|
|
"No stored initial password. It was either set explicitly at install time, or the "
|
|
"admin password has already been changed. Use Reset Password to set a new one."
|
|
)
|
|
return
|
|
with open(path) as f:
|
|
pw = f.read().strip()
|
|
print(pw if pw else "(the recorded initial password is empty)")
|
|
|
|
|
|
def list_users(args: argparse.Namespace) -> None:
|
|
"""Print all user accounts."""
|
|
run_migrations()
|
|
|
|
with Session(engine) as session:
|
|
users = session.exec(select(User).order_by(User.role, User.name)).all() # type: ignore[arg-type]
|
|
if not users:
|
|
print("No users.")
|
|
return
|
|
print(f"{'USERNAME':<20} {'NAME':<24} {'ROLE':<14} {'STATUS':<10} EMAIL")
|
|
print("-" * 84)
|
|
for u in users:
|
|
status = "active" if u.is_active else "disabled"
|
|
tag = " [SERVICE ADMIN]" if u.is_service_admin else ""
|
|
print(f"{u.username:<20} {u.name:<24} {u.role.value:<14} {status:<10} {u.email or '-'}{tag}")
|
|
print(f"\n{len(users)} user(s). The Service Admin cannot be deleted.")
|
|
|
|
|
|
def delete_user(args: argparse.Namespace) -> None:
|
|
"""Delete a user account and its dependent rows. The Service Admin is protected."""
|
|
run_migrations()
|
|
|
|
from ten31portal.routers.user_router import delete_user_cascade
|
|
|
|
with Session(engine) as session:
|
|
user = _find_user(session, args.username, args.email)
|
|
if user is None:
|
|
who = args.username or args.email
|
|
print(f"Error: no user found for '{who}'.", file=sys.stderr)
|
|
sys.exit(1)
|
|
if user.is_service_admin:
|
|
print(f"Error: '{user.username}' is the Service Admin and cannot be deleted.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
name, username = user.name, user.username
|
|
# Don't abort on a file that won't delete, but surface it — a swallowed disk or
|
|
# permission error would silently orphan the file on the data volume.
|
|
for warning in delete_user_cascade(session, user):
|
|
print(f"Warning: {warning}", file=sys.stderr)
|
|
session.commit()
|
|
print(f"Deleted user {name} ({username}).")
|
|
|
|
|
|
def dedupe_holdings(args: argparse.Namespace) -> None:
|
|
"""Remove duplicate holdings/positions left by old double-imports (fixes inflated Invested)."""
|
|
run_migrations()
|
|
from ten31portal.models import Entity
|
|
from ten31portal.routers.import_router import dedupe_entity
|
|
|
|
with Session(engine) as session:
|
|
if args.entity_id:
|
|
entities = [e for e in [session.get(Entity, args.entity_id)] if e]
|
|
else:
|
|
entities = session.exec(select(Entity).order_by(Entity.name)).all() # type: ignore[arg-type]
|
|
total_h = total_p = 0
|
|
for ent in entities:
|
|
res = dedupe_entity(ent.id, session)
|
|
if res["removed_holdings"] or res["removed_positions"]:
|
|
print(f"{ent.name}: removed {res['removed_holdings']} holding(s), "
|
|
f"{res['removed_positions']} position(s)")
|
|
total_h += res["removed_holdings"]
|
|
total_p += res["removed_positions"]
|
|
session.commit()
|
|
if total_h or total_p:
|
|
print(f"\nDone. Removed {total_h} duplicate holding(s) and {total_p} position(s).")
|
|
else:
|
|
print("No duplicates found — nothing to clean up.")
|
|
|
|
|
|
def reset_holdings(args: argparse.Namespace) -> None:
|
|
"""Clear a fund's holdings/positions/rounds so it can be re-imported from scratch."""
|
|
run_migrations()
|
|
from ten31portal.models import Entity
|
|
from ten31portal.routers.import_router import reset_entity_holdings
|
|
|
|
with Session(engine) as session:
|
|
entity = None
|
|
if args.entity_id:
|
|
entity = session.get(Entity, args.entity_id)
|
|
elif args.name:
|
|
entity = session.exec(select(Entity).where(Entity.name == args.name)).first()
|
|
if entity is None:
|
|
print(f"Error: no fund found for '{args.name or args.entity_id}'. "
|
|
f"Check the exact name with list-funds.", file=sys.stderr)
|
|
sys.exit(1)
|
|
res = reset_entity_holdings(entity.id, session)
|
|
session.commit()
|
|
print(f"Cleared {entity.name}: removed {res['holdings']} holding(s), "
|
|
f"{res['positions']} position(s), {res['rounds']} round(s). "
|
|
f"Re-import the fund's NAV to repopulate it.")
|
|
|
|
|
|
def reset_partners(args: argparse.Namespace) -> None:
|
|
"""Remove all partners (capital-account statements + access grants) from one fund."""
|
|
run_migrations()
|
|
from ten31portal.models import Entity
|
|
from ten31portal.routers.capital_import_router import reset_entity_partners
|
|
|
|
with Session(engine) as session:
|
|
entity = None
|
|
if args.entity_id:
|
|
entity = session.get(Entity, args.entity_id)
|
|
elif args.name:
|
|
entity = session.exec(select(Entity).where(Entity.name == args.name)).first()
|
|
if entity is None:
|
|
print(f"Error: no fund found for '{args.name or args.entity_id}'. "
|
|
f"Check the exact name with list-funds.", file=sys.stderr)
|
|
sys.exit(1)
|
|
res = reset_entity_partners(entity.id, session)
|
|
session.commit()
|
|
print(f"Cleared partners from {entity.name}: removed {res['statements']} capital "
|
|
f"statement(s) and {res['access_grants']} access grant(s). "
|
|
f"Investor accounts were kept. Re-import the correct roster to repopulate.")
|
|
|
|
|
|
def list_funds(args: argparse.Namespace) -> None:
|
|
"""Print every entity's id and name (so the exact name is known for reset-holdings)."""
|
|
run_migrations()
|
|
from ten31portal.models import Entity
|
|
|
|
with Session(engine) as session:
|
|
entities = session.exec(select(Entity).order_by(Entity.name)).all() # type: ignore[arg-type]
|
|
if not entities:
|
|
print("No funds yet.")
|
|
return
|
|
for e in entities:
|
|
print(f"[{e.id}] {e.name} ({e.type.value})")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(prog="ten31portal-cli")
|
|
sub = parser.add_subparsers(dest="command")
|
|
|
|
create = sub.add_parser("create-user", help="Provision a new user")
|
|
create.add_argument("--name", required=True)
|
|
create.add_argument("--username", required=True)
|
|
create.add_argument("--email", required=False, default=None)
|
|
create.add_argument("--role", required=True, choices=[r.value for r in UserRole])
|
|
create.add_argument("--password", required=True)
|
|
create.add_argument("--service-admin", action="store_true",
|
|
help="Mark as the protected built-in Service Admin")
|
|
|
|
reset = sub.add_parser("reset-password", help="Reset a user's password")
|
|
reset.add_argument("--username", required=False, default=None)
|
|
reset.add_argument("--email", required=False, default=None)
|
|
reset.add_argument("--password", required=True)
|
|
|
|
r2fa = sub.add_parser("reset-2fa", help="Clear a user's two-factor enrollment (lost phone)")
|
|
r2fa.add_argument("--username", required=False, default=None)
|
|
r2fa.add_argument("--email", required=False, default=None)
|
|
|
|
sub.add_parser("list-users", help="List all user accounts")
|
|
|
|
sub.add_parser(
|
|
"enable-investor-logins",
|
|
help="Set every no-login investor account to the default password and enable sign-in",
|
|
)
|
|
|
|
sub.add_parser("show-admin-password", help="Show the initial admin password from first boot")
|
|
|
|
delete = sub.add_parser("delete-user", help="Delete a user (not the Service Admin)")
|
|
delete.add_argument("--username", required=False, default=None)
|
|
delete.add_argument("--email", required=False, default=None)
|
|
|
|
dedupe = sub.add_parser("dedupe-holdings", help="Remove duplicate holdings/positions")
|
|
dedupe.add_argument("--entity-id", type=int, required=False, default=None,
|
|
help="Limit to one entity; omit to clean all")
|
|
|
|
sub.add_parser("list-funds", help="List entities (funds/SPVs) with their ids")
|
|
|
|
reset = sub.add_parser("reset-holdings", help="Clear a fund's holdings to re-import fresh")
|
|
reset.add_argument("--name", required=False, default=None, help="Exact fund name")
|
|
reset.add_argument("--entity-id", type=int, required=False, default=None)
|
|
|
|
rparts = sub.add_parser("reset-partners", help="Remove all partners (capital accounts + access) from a fund")
|
|
rparts.add_argument("--name", required=False, default=None, help="Exact fund name")
|
|
rparts.add_argument("--entity-id", type=int, required=False, default=None)
|
|
|
|
args = parser.parse_args()
|
|
if args.command == "create-user":
|
|
create_user(args)
|
|
elif args.command == "reset-password":
|
|
reset_password(args)
|
|
elif args.command == "reset-2fa":
|
|
reset_2fa(args)
|
|
elif args.command == "list-users":
|
|
list_users(args)
|
|
elif args.command == "enable-investor-logins":
|
|
enable_investor_logins(args)
|
|
elif args.command == "show-admin-password":
|
|
show_admin_password(args)
|
|
elif args.command == "delete-user":
|
|
delete_user(args)
|
|
elif args.command == "dedupe-holdings":
|
|
dedupe_holdings(args)
|
|
elif args.command == "list-funds":
|
|
list_funds(args)
|
|
elif args.command == "reset-holdings":
|
|
reset_holdings(args)
|
|
elif args.command == "reset-partners":
|
|
reset_partners(args)
|
|
else:
|
|
parser.print_help()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|