feat: Armarium v1.1.0 — dashboard, auth, 2FA, SMTP, settings, deploy
Dashboard: - ApexCharts bar chart (income vs fixed costs vs expenses) and donut chart - KPI cards: income, fixed costs, savings rate with configurable goal - Greeting with time-of-day and locale-aware date/time display Authentication & security: - Email-based login (no username), case-insensitive lookup - JWT access/refresh tokens with rotation and blacklist - TOTP 2FA with QR code, backup codes (copy + PDF export) - 2FA recovery via email code - Cloudflare Turnstile CAPTCHA on login and register Email flows: - Email verification on registration (24h token) - Password reset flow (15min token, anti-enumeration) - Brevo SMTP integration with HTML + plaintext email templates - Notification emails: 2FA recovery, password changed, email changed Settings page: - 2FA management (enable/disable, QR, backup codes) - Active sessions list with per-device revoke - Data export: ZIP with 6 PDFs via fpdf2 - Notification preferences (3 toggles) - Danger zone: account deletion with mandatory export + confirmation phrase UI & layout: - Sidebar with collapsible/flyout mode, Angular signal-based dropdowns - Dark mode (class-based), language switcher (DE/FR/IT/EN) - Mobile-responsive layout with touch-friendly targets - Roboto font via @fontsource (GDPR-compliant, no Google CDN) - Pure Tailwind CSS v3 Infrastructure: - Forgejo Actions CI/CD pipeline (auto-deploy on push to main) - Gunicorn + Nginx + PostgreSQL production setup - Rate limiting, HSTS, secure cookies, CSRF protection
This commit is contained in:
Daniel Krähenbühl
@@ -0,0 +1,5 @@
|
||||
from django.contrib import admin
|
||||
from .models import Account, Transaction
|
||||
|
||||
admin.site.register(Account)
|
||||
admin.site.register(Transaction)
|
||||
@@ -0,0 +1,5 @@
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class FinanceConfig(AppConfig):
|
||||
name = 'finance'
|
||||
@@ -0,0 +1,14 @@
|
||||
from django.contrib.auth.backends import ModelBackend
|
||||
from django.contrib.auth import get_user_model
|
||||
|
||||
|
||||
class EmailAuthBackend(ModelBackend):
|
||||
def authenticate(self, request, username=None, password=None, **kwargs):
|
||||
User = get_user_model()
|
||||
try:
|
||||
user = User.objects.get(email__iexact=username)
|
||||
except (User.DoesNotExist, User.MultipleObjectsReturned):
|
||||
return None
|
||||
if user.check_password(password) and self.user_can_authenticate(user):
|
||||
return user
|
||||
return None
|
||||
@@ -0,0 +1,33 @@
|
||||
import logging
|
||||
from django.conf import settings
|
||||
from django.core.mail import EmailMultiAlternatives
|
||||
from django.template.loader import render_to_string
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def send_email(template_name: str, context: dict, subject: str, to: str | list[str]) -> bool:
|
||||
"""
|
||||
Render and send an HTML email with plaintext fallback.
|
||||
Returns True on success, False on failure.
|
||||
"""
|
||||
if isinstance(to, str):
|
||||
to = [to]
|
||||
|
||||
html = render_to_string(f'emails/{template_name}.html', context)
|
||||
text = render_to_string(f'emails/{template_name}.txt', context)
|
||||
|
||||
msg = EmailMultiAlternatives(
|
||||
subject=subject,
|
||||
body=text,
|
||||
from_email=settings.DEFAULT_FROM_EMAIL,
|
||||
to=to,
|
||||
)
|
||||
msg.attach_alternative(html, 'text/html')
|
||||
|
||||
try:
|
||||
msg.send()
|
||||
return True
|
||||
except Exception:
|
||||
logger.exception('Failed to send email "%s" to %s', template_name, to)
|
||||
return False
|
||||
@@ -0,0 +1,25 @@
|
||||
# Generated by Django 6.0.3 on 2026-03-08 17:03
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
initial = True
|
||||
|
||||
dependencies = [
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='Account',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name', models.CharField(max_length=100)),
|
||||
('account_type', models.CharField(choices=[('asset', 'Asset Account (Bank/Cash)'), ('expense', 'Expense Account (Laden/Empfänger)'), ('revenue', 'Revenue Account (Einnahmequelle)')], default='asset', max_length=20)),
|
||||
('balance', models.DecimalField(decimal_places=2, default=0.0, max_digits=12)),
|
||||
('active', models.BooleanField(default=True)),
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
],
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,25 @@
|
||||
# Generated by Django 6.0.3 on 2026-03-08 17:11
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0001_initial'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='Transaction',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('description', models.CharField(max_length=255)),
|
||||
('amount', models.DecimalField(decimal_places=2, max_digits=12)),
|
||||
('date', models.DateField()),
|
||||
('destination_account', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='deposits', to='finance.account')),
|
||||
('source_account', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='withdrawals', to='finance.account')),
|
||||
],
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,26 @@
|
||||
# Generated by Django 6.0.3 on 2026-03-15 16:10
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0002_transaction'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='Budget',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name', models.CharField(max_length=100)),
|
||||
('amount', models.DecimalField(decimal_places=2, max_digits=12)),
|
||||
('category', models.CharField(choices=[('housing', 'Wohnen'), ('food', 'Lebensmittel'), ('transport', 'Transport'), ('health', 'Gesundheit'), ('entertainment', 'Freizeit'), ('savings', 'Sparen'), ('other', 'Sonstiges')], default='other', max_length=50)),
|
||||
('due_day', models.PositiveSmallIntegerField(help_text='Tag im Monat (1-31)')),
|
||||
('active', models.BooleanField(default=True)),
|
||||
('account', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='budgets', to='finance.account')),
|
||||
],
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,37 @@
|
||||
import django.db.models.deletion
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0003_budget'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.RemoveField(
|
||||
model_name='budget',
|
||||
name='due_day',
|
||||
),
|
||||
migrations.RenameField(
|
||||
model_name='budget',
|
||||
old_name='category',
|
||||
new_name='main_category',
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name='budget',
|
||||
name='main_category',
|
||||
field=models.CharField(
|
||||
choices=[
|
||||
('fixed_expenses', 'Fixe Ausgaben'),
|
||||
('mobile_internet', 'Mobile & Internet'),
|
||||
('leisure', 'Freizeit'),
|
||||
('tax_reserves', 'Steuerrücklagen'),
|
||||
('insurance', 'Versicherungen'),
|
||||
('loans', 'Abzahlungen & Kredite'),
|
||||
],
|
||||
default='fixed_expenses',
|
||||
max_length=50,
|
||||
),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,18 @@
|
||||
# Generated by Django 6.0.3 on 2026-03-23 22:01
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0004_alter_budget'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AlterField(
|
||||
model_name='budget',
|
||||
name='main_category',
|
||||
field=models.CharField(choices=[('fixed_expenses', 'Fixe Ausgaben'), ('mobile_internet', 'Mobile & Internet'), ('subscriptions', 'Abonnements'), ('leisure', 'Freizeit'), ('tax_reserves', 'Steuerrücklagen'), ('insurance', 'Versicherungen'), ('loans', 'Abzahlungen & Kredite')], default='fixed_expenses', max_length=50),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,26 @@
|
||||
# Generated by Django 6.0.3 on 2026-03-23 22:10
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0005_add_subscriptions_category'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='Expense',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name', models.CharField(max_length=255)),
|
||||
('amount', models.DecimalField(decimal_places=2, max_digits=12)),
|
||||
('date', models.DateField()),
|
||||
('category', models.CharField(choices=[('groceries', 'Groceries'), ('dining', 'Dining & Restaurants'), ('transport', 'Transport'), ('health', 'Health & Medical'), ('clothing', 'Clothing'), ('electronics', 'Electronics'), ('household', 'Household'), ('entertainment', 'Entertainment'), ('travel', 'Travel'), ('other', 'Other')], default='other', max_length=50)),
|
||||
('notes', models.TextField(blank=True, default='')),
|
||||
('account', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='expenses', to='finance.account')),
|
||||
],
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,23 @@
|
||||
# Generated by Django 6.0.3 on 2026-03-24 19:17
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0006_add_expense_model'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='Profile',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('first_name', models.CharField(blank=True, default='', max_length=100)),
|
||||
('last_name', models.CharField(blank=True, default='', max_length=100)),
|
||||
('email', models.EmailField(blank=True, default='', max_length=254)),
|
||||
('avatar_color', models.CharField(default='#1A56DB', max_length=7)),
|
||||
],
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,18 @@
|
||||
# Generated by Django 6.0.3 on 2026-03-24 19:28
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0007_add_profile_model'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='avatar_image',
|
||||
field=models.ImageField(blank=True, null=True, upload_to='avatars/'),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,44 @@
|
||||
# Generated by Django 6.0.3 on 2026-03-24 19:38
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
def assign_legacy_user(apps, schema_editor):
|
||||
User = apps.get_model('auth', 'User')
|
||||
Account = apps.get_model('finance', 'Account')
|
||||
Profile = apps.get_model('finance', 'Profile')
|
||||
|
||||
legacy_user, _ = User.objects.get_or_create(
|
||||
username='legacy_user',
|
||||
defaults={'email': '', 'is_active': True},
|
||||
)
|
||||
|
||||
Account.objects.filter(user__isnull=True).update(user=legacy_user)
|
||||
|
||||
for profile in Profile.objects.filter(user__isnull=True):
|
||||
profile.user = legacy_user
|
||||
profile.save()
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0008_add_avatar_image'),
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='account',
|
||||
name='user',
|
||||
field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='accounts', to=settings.AUTH_USER_MODEL),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='user',
|
||||
field=models.OneToOneField(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='profile', to=settings.AUTH_USER_MODEL),
|
||||
),
|
||||
migrations.RunPython(assign_legacy_user, migrations.RunPython.noop),
|
||||
]
|
||||
@@ -0,0 +1,37 @@
|
||||
# Generated by Django 6.0.3 on 2026-03-24 20:57
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0009_add_user_fk'),
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='expense',
|
||||
name='due_date',
|
||||
field=models.DateField(blank=True, null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='canton',
|
||||
field=models.CharField(choices=[('AG', 'Aargau'), ('AI', 'Appenzell Innerrhoden'), ('AR', 'Appenzell Ausserrhoden'), ('BE', 'Bern'), ('BL', 'Basel-Landschaft'), ('BS', 'Basel-Stadt'), ('FR', 'Fribourg'), ('GE', 'Geneva'), ('GL', 'Glarus'), ('GR', 'Graubünden'), ('JU', 'Jura'), ('LU', 'Lucerne'), ('NE', 'Neuchâtel'), ('NW', 'Nidwalden'), ('OW', 'Obwalden'), ('SG', 'St. Gallen'), ('SH', 'Schaffhausen'), ('SO', 'Solothurn'), ('SZ', 'Schwyz'), ('TG', 'Thurgau'), ('TI', 'Ticino'), ('UR', 'Uri'), ('VD', 'Vaud'), ('VS', 'Valais'), ('ZG', 'Zug'), ('ZH', 'Zurich')], default='ZH', max_length=2),
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='Deadline',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('title', models.CharField(max_length=200)),
|
||||
('date', models.DateField()),
|
||||
('type', models.CharField(choices=[('tax', 'Tax'), ('insurance', 'Insurance'), ('invoice', 'Invoice'), ('personal', 'Personal'), ('other', 'Other')], default='other', max_length=20)),
|
||||
('notes', models.TextField(blank=True, default='')),
|
||||
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='deadlines', to=settings.AUTH_USER_MODEL)),
|
||||
],
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,28 @@
|
||||
# Generated by Django 6.0.4 on 2026-04-12 18:25
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0010_add_calendar_fields'),
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='ReadEvent',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('event_type', models.CharField(choices=[('deadline', 'Deadline'), ('expense', 'Expense')], max_length=20)),
|
||||
('event_id', models.PositiveIntegerField()),
|
||||
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='read_events', to=settings.AUTH_USER_MODEL)),
|
||||
],
|
||||
options={
|
||||
'unique_together': {('user', 'event_type', 'event_id')},
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,18 @@
|
||||
# Generated by Django 6.0.4 on 2026-04-12 18:55
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0011_readevent'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='language',
|
||||
field=models.CharField(choices=[('de', 'Deutsch'), ('fr', 'Français'), ('it', 'Italiano'), ('en', 'English')], default='de', max_length=2),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,21 @@
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0012_profile_language'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='totp_secret',
|
||||
field=models.CharField(blank=True, default='', max_length=64),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='totp_enabled',
|
||||
field=models.BooleanField(default=False),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,36 @@
|
||||
from django.db import migrations, models
|
||||
import django.db.models.deletion
|
||||
from django.conf import settings
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0013_profile_totp'),
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='totp_last_used_code',
|
||||
field=models.CharField(blank=True, default='', max_length=6),
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='BackupCode',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('code_hash', models.CharField(max_length=64)),
|
||||
('used', models.BooleanField(default=False)),
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
('user', models.ForeignKey(
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name='backup_codes',
|
||||
to=settings.AUTH_USER_MODEL,
|
||||
)),
|
||||
],
|
||||
options={
|
||||
'indexes': [models.Index(fields=['user', 'used'], name='finance_bac_user_id_idx')],
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,16 @@
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0014_totp_security'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='recovery_email',
|
||||
field=models.EmailField(blank=True, default=''),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,21 @@
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0015_profile_recovery_email'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='recovery_code_hash',
|
||||
field=models.CharField(blank=True, default='', max_length=64),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='recovery_code_expires',
|
||||
field=models.DateTimeField(blank=True, null=True),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,47 @@
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
import django.db.models.deletion
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0016_profile_recovery_code'),
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='notif_deadlines',
|
||||
field=models.BooleanField(default=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='notif_budget_alerts',
|
||||
field=models.BooleanField(default=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='notif_monthly_summary',
|
||||
field=models.BooleanField(default=False),
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='UserSession',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('session_key', models.CharField(max_length=64, unique=True)),
|
||||
('refresh_jti', models.CharField(blank=True, default='', max_length=255)),
|
||||
('device_name', models.CharField(blank=True, default='', max_length=200)),
|
||||
('ip_address', models.GenericIPAddressField(blank=True, null=True)),
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
('last_active_at', models.DateTimeField(auto_now_add=True)),
|
||||
('user', models.ForeignKey(
|
||||
on_delete=django.db.models.deletion.CASCADE,
|
||||
related_name='user_sessions',
|
||||
to=settings.AUTH_USER_MODEL,
|
||||
)),
|
||||
],
|
||||
options={'ordering': ['-last_active_at']},
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,28 @@
|
||||
# Generated by Django 6.0.4 on 2026-05-18 19:22
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0017_user_sessions_notifications'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.RenameIndex(
|
||||
model_name='backupcode',
|
||||
new_name='finance_bac_user_id_7e357d_idx',
|
||||
old_name='finance_bac_user_id_idx',
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='savings_rate_goal',
|
||||
field=models.PositiveSmallIntegerField(default=20),
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name='profile',
|
||||
name='canton',
|
||||
field=models.CharField(choices=[('AG', 'Aargau'), ('AI', 'Appenzell Innerrhoden'), ('AR', 'Appenzell Ausserrhoden'), ('BE', 'Bern'), ('BL', 'Basel-Landschaft'), ('BS', 'Basel-Stadt'), ('FR', 'Fribourg'), ('GE', 'Geneva'), ('GL', 'Glarus'), ('GR', 'Graubünden'), ('JU', 'Jura'), ('LU', 'Lucerne'), ('NE', 'Neuchâtel'), ('NW', 'Nidwalden'), ('OW', 'Obwalden'), ('SG', 'St. Gallen'), ('SH', 'Schaffhausen'), ('SO', 'Solothurn'), ('SZ', 'Schwyz'), ('TG', 'Thurgau'), ('TI', 'Ticino'), ('UR', 'Uri'), ('VD', 'Vaud'), ('VS', 'Valais'), ('ZG', 'Zug'), ('ZH', 'Zürich')], default='ZH', max_length=2),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,33 @@
|
||||
# Generated by Django 6.0.4 on 2026-05-19 17:54
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0018_profile_savings_rate_goal'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='email_verified',
|
||||
field=models.BooleanField(default=False),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='email_verify_token',
|
||||
field=models.CharField(blank=True, default='', max_length=64),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='password_reset_token_expires',
|
||||
field=models.DateTimeField(blank=True, null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='password_reset_token_hash',
|
||||
field=models.CharField(blank=True, default='', max_length=64),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,18 @@
|
||||
# Generated by Django 6.0.4 on 2026-05-19 18:11
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('finance', '0019_profile_email_verification_and_password_reset'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='profile',
|
||||
name='email_verify_token_expires',
|
||||
field=models.DateTimeField(blank=True, null=True),
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,226 @@
|
||||
from django.db import models
|
||||
from django.conf import settings
|
||||
|
||||
class Account(models.Model):
|
||||
# Typen basierend auf der Firefly III Logik
|
||||
ACCOUNT_TYPES = [
|
||||
('asset', 'Asset Account (Bank/Cash)'),
|
||||
('expense', 'Expense Account (Laden/Empfänger)'),
|
||||
('revenue', 'Revenue Account (Einnahmequelle)'),
|
||||
]
|
||||
|
||||
user = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='accounts',
|
||||
null=True,
|
||||
)
|
||||
name = models.CharField(max_length=100)
|
||||
account_type = models.CharField(max_length=20, choices=ACCOUNT_TYPES, default='asset')
|
||||
balance = models.DecimalField(max_digits=12, decimal_places=2, default=0.00)
|
||||
active = models.BooleanField(default=True)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.name} ({self.get_account_type_display()})"
|
||||
|
||||
class Transaction(models.Model):
|
||||
description = models.CharField(max_length=255)
|
||||
amount = models.DecimalField(max_digits=12, decimal_places=2)
|
||||
date = models.DateField()
|
||||
|
||||
# Die Verknüpfung zu den Konten (Double-Entry Prinzip)
|
||||
source_account = models.ForeignKey(
|
||||
Account,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='withdrawals'
|
||||
)
|
||||
destination_account = models.ForeignKey(
|
||||
Account,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='deposits'
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.date}: {self.description} ({self.amount}€)"
|
||||
|
||||
|
||||
class Budget(models.Model):
|
||||
MAIN_CATEGORY_CHOICES = [
|
||||
('fixed_expenses', 'Fixe Ausgaben'),
|
||||
('mobile_internet', 'Mobile & Internet'),
|
||||
('subscriptions', 'Abonnements'),
|
||||
('leisure', 'Freizeit'),
|
||||
('tax_reserves', 'Steuerrücklagen'),
|
||||
('insurance', 'Versicherungen'),
|
||||
('loans', 'Abzahlungen & Kredite'),
|
||||
]
|
||||
|
||||
name = models.CharField(max_length=100)
|
||||
amount = models.DecimalField(max_digits=12, decimal_places=2)
|
||||
main_category = models.CharField(max_length=50, choices=MAIN_CATEGORY_CHOICES, default='fixed_expenses')
|
||||
account = models.ForeignKey(
|
||||
Account,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='budgets'
|
||||
)
|
||||
active = models.BooleanField(default=True)
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.name} ({self.amount} CHF)"
|
||||
|
||||
|
||||
class Expense(models.Model):
|
||||
CATEGORY_CHOICES = [
|
||||
('groceries', 'Groceries'),
|
||||
('dining', 'Dining & Restaurants'),
|
||||
('transport', 'Transport'),
|
||||
('health', 'Health & Medical'),
|
||||
('clothing', 'Clothing'),
|
||||
('electronics', 'Electronics'),
|
||||
('household', 'Household'),
|
||||
('entertainment', 'Entertainment'),
|
||||
('travel', 'Travel'),
|
||||
('other', 'Other'),
|
||||
]
|
||||
|
||||
name = models.CharField(max_length=255)
|
||||
amount = models.DecimalField(max_digits=12, decimal_places=2)
|
||||
date = models.DateField()
|
||||
category = models.CharField(max_length=50, choices=CATEGORY_CHOICES, default='other')
|
||||
account = models.ForeignKey(
|
||||
Account,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='expenses'
|
||||
)
|
||||
notes = models.TextField(blank=True, default='')
|
||||
due_date = models.DateField(blank=True, null=True)
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.date}: {self.name} ({self.amount} CHF)"
|
||||
|
||||
|
||||
CANTON_CHOICES = [
|
||||
('AG', 'Aargau'), ('AI', 'Appenzell Innerrhoden'), ('AR', 'Appenzell Ausserrhoden'),
|
||||
('BE', 'Bern'), ('BL', 'Basel-Landschaft'), ('BS', 'Basel-Stadt'),
|
||||
('FR', 'Fribourg'), ('GE', 'Geneva'), ('GL', 'Glarus'),
|
||||
('GR', 'Graubünden'), ('JU', 'Jura'), ('LU', 'Lucerne'),
|
||||
('NE', 'Neuchâtel'), ('NW', 'Nidwalden'), ('OW', 'Obwalden'),
|
||||
('SG', 'St. Gallen'), ('SH', 'Schaffhausen'), ('SO', 'Solothurn'),
|
||||
('SZ', 'Schwyz'), ('TG', 'Thurgau'), ('TI', 'Ticino'),
|
||||
('UR', 'Uri'), ('VD', 'Vaud'), ('VS', 'Valais'),
|
||||
('ZG', 'Zug'), ('ZH', 'Zürich'),
|
||||
]
|
||||
|
||||
|
||||
class Deadline(models.Model):
|
||||
TYPE_CHOICES = [
|
||||
('tax', 'Tax'),
|
||||
('insurance', 'Insurance'),
|
||||
('invoice', 'Invoice'),
|
||||
('personal', 'Personal'),
|
||||
('other', 'Other'),
|
||||
]
|
||||
user = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='deadlines',
|
||||
)
|
||||
title = models.CharField(max_length=200)
|
||||
date = models.DateField()
|
||||
type = models.CharField(max_length=20, choices=TYPE_CHOICES, default='other')
|
||||
notes = models.TextField(blank=True, default='')
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.date}: {self.title}"
|
||||
|
||||
|
||||
class Profile(models.Model):
|
||||
user = models.OneToOneField(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='profile',
|
||||
null=True,
|
||||
)
|
||||
first_name = models.CharField(max_length=100, blank=True, default='')
|
||||
last_name = models.CharField(max_length=100, blank=True, default='')
|
||||
email = models.EmailField(blank=True, default='')
|
||||
avatar_color = models.CharField(max_length=7, default='#1A56DB')
|
||||
avatar_image = models.ImageField(upload_to='avatars/', blank=True, null=True)
|
||||
canton = models.CharField(max_length=2, choices=CANTON_CHOICES, default='ZH')
|
||||
language = models.CharField(max_length=2, choices=[('de','Deutsch'),('fr','Français'),('it','Italiano'),('en','English')], default='de')
|
||||
totp_secret = models.CharField(max_length=64, blank=True, default='')
|
||||
totp_enabled = models.BooleanField(default=False)
|
||||
totp_last_used_code = models.CharField(max_length=6, blank=True, default='')
|
||||
recovery_email = models.EmailField(blank=True, default='')
|
||||
recovery_code_hash = models.CharField(max_length=64, blank=True, default='')
|
||||
recovery_code_expires = models.DateTimeField(null=True, blank=True)
|
||||
notif_deadlines = models.BooleanField(default=True)
|
||||
notif_budget_alerts = models.BooleanField(default=True)
|
||||
notif_monthly_summary = models.BooleanField(default=False)
|
||||
savings_rate_goal = models.PositiveSmallIntegerField(default=20)
|
||||
email_verified = models.BooleanField(default=False)
|
||||
email_verify_token = models.CharField(max_length=64, blank=True, default='')
|
||||
email_verify_token_expires = models.DateTimeField(null=True, blank=True)
|
||||
password_reset_token_hash = models.CharField(max_length=64, blank=True, default='')
|
||||
password_reset_token_expires = models.DateTimeField(null=True, blank=True)
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.first_name} {self.last_name}".strip() or 'Profile'
|
||||
|
||||
|
||||
class UserSession(models.Model):
|
||||
user = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='user_sessions',
|
||||
)
|
||||
session_key = models.CharField(max_length=64, unique=True)
|
||||
refresh_jti = models.CharField(max_length=255, blank=True, default='')
|
||||
device_name = models.CharField(max_length=200, blank=True, default='')
|
||||
ip_address = models.GenericIPAddressField(null=True, blank=True)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
last_active_at = models.DateTimeField(auto_now_add=True)
|
||||
|
||||
class Meta:
|
||||
ordering = ['-last_active_at']
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.user} – {self.device_name}"
|
||||
|
||||
|
||||
class ReadEvent(models.Model):
|
||||
EVENT_TYPES = [
|
||||
('deadline', 'Deadline'),
|
||||
('expense', 'Expense'),
|
||||
]
|
||||
user = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='read_events',
|
||||
)
|
||||
event_type = models.CharField(max_length=20, choices=EVENT_TYPES)
|
||||
event_id = models.PositiveIntegerField()
|
||||
|
||||
class Meta:
|
||||
unique_together = ['user', 'event_type', 'event_id']
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.user} – {self.event_type} {self.event_id}"
|
||||
|
||||
|
||||
class BackupCode(models.Model):
|
||||
user = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='backup_codes',
|
||||
)
|
||||
code_hash = models.CharField(max_length=64)
|
||||
used = models.BooleanField(default=False)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
|
||||
class Meta:
|
||||
indexes = [models.Index(fields=['user', 'used'])]
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.user} – backup {'used' if self.used else 'active'}"
|
||||
@@ -0,0 +1,74 @@
|
||||
from rest_framework import serializers
|
||||
from django.contrib.auth import get_user_model
|
||||
from .models import Account, Transaction, Budget, Expense, Profile, Deadline
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
class AccountSerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = Account
|
||||
exclude = ['user']
|
||||
|
||||
|
||||
class TransactionSerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = Transaction
|
||||
fields = '__all__'
|
||||
|
||||
def validate(self, data):
|
||||
request = self.context.get('request')
|
||||
if not request:
|
||||
return data
|
||||
user = request.user
|
||||
source = data.get('source_account') or (self.instance.source_account if self.instance else None)
|
||||
dest = data.get('destination_account') or (self.instance.destination_account if self.instance else None)
|
||||
if source and source.user != user:
|
||||
raise serializers.ValidationError('Source account does not belong to you.')
|
||||
if dest and dest.user != user:
|
||||
raise serializers.ValidationError('Destination account does not belong to you.')
|
||||
return data
|
||||
|
||||
|
||||
class BudgetSerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = Budget
|
||||
fields = '__all__'
|
||||
|
||||
|
||||
class ExpenseSerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = Expense
|
||||
fields = '__all__'
|
||||
|
||||
|
||||
class ProfileSerializer(serializers.ModelSerializer):
|
||||
totp_enabled = serializers.BooleanField(read_only=True)
|
||||
|
||||
class Meta:
|
||||
model = Profile
|
||||
exclude = ['user', 'totp_secret', 'email_verify_token', 'email_verify_token_expires', 'password_reset_token_hash', 'password_reset_token_expires']
|
||||
|
||||
|
||||
class DeadlineSerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = Deadline
|
||||
exclude = ['user']
|
||||
|
||||
|
||||
class RegisterSerializer(serializers.Serializer):
|
||||
email = serializers.EmailField()
|
||||
password = serializers.CharField(min_length=8, write_only=True)
|
||||
|
||||
def validate_email(self, value):
|
||||
if User.objects.filter(email=value).exists():
|
||||
raise serializers.ValidationError('Email already registered.')
|
||||
return value
|
||||
|
||||
def create(self, validated_data):
|
||||
email = validated_data['email']
|
||||
return User.objects.create_user(
|
||||
username=email,
|
||||
email=email,
|
||||
password=validated_data['password'],
|
||||
)
|
||||
@@ -0,0 +1,3 @@
|
||||
from django.test import TestCase
|
||||
|
||||
# Create your tests here.
|
||||
@@ -0,0 +1,999 @@
|
||||
import base64
|
||||
import hmac
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import secrets
|
||||
import time
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
import pyotp
|
||||
|
||||
logger = logging.getLogger('armarium')
|
||||
from django.conf import settings
|
||||
from django.contrib.auth import get_user_model, authenticate
|
||||
from django.http import HttpResponse
|
||||
from icalendar import Calendar as iCalendar, Event as iCalEvent
|
||||
|
||||
from rest_framework import viewsets, views, status
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.permissions import AllowAny, IsAuthenticated
|
||||
from rest_framework.throttling import AnonRateThrottle
|
||||
from rest_framework_simplejwt.tokens import RefreshToken
|
||||
from rest_framework_simplejwt.exceptions import TokenError
|
||||
from .models import Account, Transaction, Budget, Expense, Profile, Deadline, ReadEvent, BackupCode, UserSession
|
||||
from .serializers import (
|
||||
AccountSerializer, TransactionSerializer, BudgetSerializer,
|
||||
ExpenseSerializer, ProfileSerializer, DeadlineSerializer, RegisterSerializer,
|
||||
)
|
||||
|
||||
|
||||
def _verify_turnstile(token: str, remote_ip: str = '') -> bool:
|
||||
if settings.DEBUG:
|
||||
return True
|
||||
if not token or not settings.TURNSTILE_SECRET_KEY:
|
||||
return False
|
||||
data = urllib.parse.urlencode({
|
||||
'secret': settings.TURNSTILE_SECRET_KEY,
|
||||
'response': token,
|
||||
'remoteip': remote_ip,
|
||||
}).encode()
|
||||
try:
|
||||
req = urllib.request.Request(
|
||||
'https://challenges.cloudflare.com/turnstile/v0/siteverify',
|
||||
data=data,
|
||||
method='POST',
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||||
return json.loads(resp.read()).get('success', False)
|
||||
except Exception:
|
||||
logger.warning('Turnstile verification request failed')
|
||||
return False
|
||||
|
||||
|
||||
def generate_ical_token(user_id: int) -> str:
|
||||
return hmac.new(
|
||||
settings.SECRET_KEY.encode(),
|
||||
str(user_id).encode(),
|
||||
hashlib.sha256
|
||||
).hexdigest()
|
||||
|
||||
MAX_AVATAR_SIZE_BYTES = 2 * 1024 * 1024 # 2 MB
|
||||
|
||||
|
||||
class AuthThrottle(AnonRateThrottle):
|
||||
rate = '5/min'
|
||||
|
||||
|
||||
class AccountViewSet(viewsets.ModelViewSet):
|
||||
serializer_class = AccountSerializer
|
||||
|
||||
def get_queryset(self):
|
||||
return Account.objects.filter(user=self.request.user)
|
||||
|
||||
def perform_create(self, serializer):
|
||||
serializer.save(user=self.request.user)
|
||||
|
||||
|
||||
class TransactionViewSet(viewsets.ModelViewSet):
|
||||
serializer_class = TransactionSerializer
|
||||
|
||||
def get_queryset(self):
|
||||
return Transaction.objects.filter(source_account__user=self.request.user)
|
||||
|
||||
def get_serializer_context(self):
|
||||
context = super().get_serializer_context()
|
||||
context['request'] = self.request
|
||||
return context
|
||||
|
||||
|
||||
class BudgetViewSet(viewsets.ModelViewSet):
|
||||
serializer_class = BudgetSerializer
|
||||
|
||||
def get_queryset(self):
|
||||
return Budget.objects.filter(account__user=self.request.user)
|
||||
|
||||
|
||||
class ExpenseViewSet(viewsets.ModelViewSet):
|
||||
serializer_class = ExpenseSerializer
|
||||
|
||||
def get_queryset(self):
|
||||
return Expense.objects.filter(account__user=self.request.user)
|
||||
|
||||
|
||||
class DeadlineViewSet(viewsets.ModelViewSet):
|
||||
serializer_class = DeadlineSerializer
|
||||
|
||||
def get_queryset(self):
|
||||
return Deadline.objects.filter(user=self.request.user).order_by('date')
|
||||
|
||||
def perform_create(self, serializer):
|
||||
serializer.save(user=self.request.user)
|
||||
|
||||
|
||||
class ProfileView(views.APIView):
|
||||
def get(self, request):
|
||||
profile, _ = Profile.objects.get_or_create(user=request.user)
|
||||
return Response(ProfileSerializer(profile).data)
|
||||
|
||||
def put(self, request):
|
||||
from .email import send_email
|
||||
|
||||
avatar = request.FILES.get('avatar_image')
|
||||
if avatar and avatar.size > MAX_AVATAR_SIZE_BYTES:
|
||||
return Response({'detail': 'Image must be smaller than 2 MB.'}, status=400)
|
||||
|
||||
recovery_email = request.data.get('recovery_email', '').strip().lower()
|
||||
if recovery_email and recovery_email == request.user.email.lower():
|
||||
return Response(
|
||||
{'recovery_email': 'Recovery email must differ from your login email.'},
|
||||
status=400,
|
||||
)
|
||||
|
||||
old_email = request.user.email
|
||||
profile, _ = Profile.objects.get_or_create(user=request.user)
|
||||
serializer = ProfileSerializer(profile, data=request.data, partial=True)
|
||||
if serializer.is_valid():
|
||||
serializer.save()
|
||||
new_email = request.user.email
|
||||
if new_email != old_email:
|
||||
send_email(
|
||||
'email_changed',
|
||||
{'new_email': new_email},
|
||||
'Armarium – Deine E-Mail-Adresse wurde geändert',
|
||||
old_email,
|
||||
)
|
||||
return Response(serializer.data)
|
||||
return Response(serializer.errors, status=400)
|
||||
|
||||
def delete(self, request):
|
||||
password = request.data.get('password', '')
|
||||
if not password or not request.user.check_password(password):
|
||||
return Response({'detail': 'Passwort ungültig.'}, status=status.HTTP_403_FORBIDDEN)
|
||||
request.user.delete()
|
||||
return Response(status=status.HTTP_204_NO_CONTENT)
|
||||
|
||||
|
||||
class RegisterView(views.APIView):
|
||||
permission_classes = [AllowAny]
|
||||
throttle_classes = [AuthThrottle]
|
||||
|
||||
def post(self, request):
|
||||
from .email import send_email
|
||||
|
||||
if not _verify_turnstile(
|
||||
request.data.get('cf_turnstile_response', ''),
|
||||
request.META.get('REMOTE_ADDR', ''),
|
||||
):
|
||||
return Response({'detail': 'Captcha verification failed.'}, status=status.HTTP_400_BAD_REQUEST)
|
||||
serializer = RegisterSerializer(data=request.data)
|
||||
if serializer.is_valid():
|
||||
user = serializer.save()
|
||||
from django.utils import timezone
|
||||
from datetime import timedelta
|
||||
token = secrets.token_urlsafe(32)
|
||||
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
||||
profile, _ = Profile.objects.get_or_create(user=user)
|
||||
profile.email_verify_token = token_hash
|
||||
profile.email_verify_token_expires = timezone.now() + timedelta(hours=24)
|
||||
profile.save(update_fields=['email_verify_token', 'email_verify_token_expires'])
|
||||
link = f"{settings.FRONTEND_URL}/verify-email?token={token}"
|
||||
send_email('registration_confirm', {'link': link}, 'Armarium – E-Mail-Adresse bestätigen', user.email)
|
||||
return Response({'detail': 'Account created.'}, status=status.HTTP_201_CREATED)
|
||||
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
|
||||
class LogoutView(views.APIView):
|
||||
permission_classes = [AllowAny]
|
||||
|
||||
def post(self, request):
|
||||
refresh_token = request.data.get('refresh')
|
||||
if not refresh_token:
|
||||
return Response({'detail': 'Refresh token required.'}, status=400)
|
||||
try:
|
||||
token = RefreshToken(refresh_token)
|
||||
jti = token.payload.get('jti', '')
|
||||
token.blacklist()
|
||||
if jti:
|
||||
UserSession.objects.filter(refresh_jti=jti).delete()
|
||||
except TokenError:
|
||||
pass # already invalid, treat as success
|
||||
|
||||
session_key = request.headers.get('X-Session-Key', '')
|
||||
if session_key:
|
||||
UserSession.objects.filter(session_key=session_key).delete()
|
||||
|
||||
return Response(status=status.HTTP_204_NO_CONTENT)
|
||||
|
||||
|
||||
class ChangePasswordView(views.APIView):
|
||||
def post(self, request):
|
||||
from .email import send_email
|
||||
|
||||
password = request.data.get('password', '')
|
||||
if len(password) < 8:
|
||||
return Response({'detail': 'Password must be at least 8 characters.'}, status=400)
|
||||
request.user.set_password(password)
|
||||
request.user.save()
|
||||
|
||||
current_key = request.headers.get('X-Session-Key', '')
|
||||
other_sessions = UserSession.objects.filter(user=request.user).exclude(session_key=current_key)
|
||||
for session in other_sessions:
|
||||
_blacklist_session(session)
|
||||
|
||||
send_email('password_changed', {}, 'Armarium – Dein Passwort wurde geändert', request.user.email)
|
||||
return Response({'detail': 'Password updated.'})
|
||||
|
||||
|
||||
class SearchView(views.APIView):
|
||||
"""Global search across all user resources."""
|
||||
|
||||
def get(self, request):
|
||||
q = request.query_params.get('q', '').strip()
|
||||
if len(q) < 2:
|
||||
return Response({})
|
||||
|
||||
user = request.user
|
||||
results = {}
|
||||
|
||||
accounts = Account.objects.filter(user=user, name__icontains=q)[:5]
|
||||
if accounts:
|
||||
results['accounts'] = [
|
||||
{'id': a.id, 'title': a.name, 'subtitle': a.get_account_type_display()}
|
||||
for a in accounts
|
||||
]
|
||||
|
||||
budgets = Budget.objects.filter(account__user=user, name__icontains=q)[:5]
|
||||
if budgets:
|
||||
results['budgets'] = [
|
||||
{'id': b.id, 'title': b.name, 'subtitle': f'CHF {b.amount}'}
|
||||
for b in budgets
|
||||
]
|
||||
|
||||
expenses = Expense.objects.filter(account__user=user, name__icontains=q)[:5]
|
||||
if expenses:
|
||||
results['expenses'] = [
|
||||
{'id': e.id, 'title': e.name, 'subtitle': f'{e.date} · CHF {e.amount}'}
|
||||
for e in expenses
|
||||
]
|
||||
|
||||
transactions = Transaction.objects.filter(
|
||||
source_account__user=user, description__icontains=q
|
||||
)[:5]
|
||||
if transactions:
|
||||
results['transactions'] = [
|
||||
{'id': t.id, 'title': t.description, 'subtitle': f'{t.date} · CHF {t.amount}'}
|
||||
for t in transactions
|
||||
]
|
||||
|
||||
deadlines = Deadline.objects.filter(user=user, title__icontains=q)[:5]
|
||||
if deadlines:
|
||||
results['deadlines'] = [
|
||||
{'id': d.id, 'title': d.title, 'subtitle': str(d.date), 'date': str(d.date)}
|
||||
for d in deadlines
|
||||
]
|
||||
|
||||
return Response(results)
|
||||
|
||||
|
||||
class NotificationsView(views.APIView):
|
||||
"""Returns all unread active events (date <= today) for the authenticated user."""
|
||||
|
||||
def get(self, request):
|
||||
from datetime import date
|
||||
today = date.today()
|
||||
|
||||
read_deadlines = set(
|
||||
ReadEvent.objects.filter(user=request.user, event_type='deadline')
|
||||
.values_list('event_id', flat=True)
|
||||
)
|
||||
read_expenses = set(
|
||||
ReadEvent.objects.filter(user=request.user, event_type='expense')
|
||||
.values_list('event_id', flat=True)
|
||||
)
|
||||
|
||||
notifications = []
|
||||
|
||||
for d in Deadline.objects.filter(user=request.user, date__lte=today):
|
||||
if d.id not in read_deadlines:
|
||||
notifications.append({
|
||||
'event_type': 'deadline',
|
||||
'event_id': d.id,
|
||||
'title': d.title,
|
||||
'date': str(d.date),
|
||||
})
|
||||
|
||||
for e in Expense.objects.filter(account__user=request.user, due_date__lte=today):
|
||||
if e.id not in read_expenses:
|
||||
notifications.append({
|
||||
'event_type': 'expense',
|
||||
'event_id': e.id,
|
||||
'title': e.name,
|
||||
'date': str(e.due_date),
|
||||
})
|
||||
|
||||
notifications.sort(key=lambda x: x['date'])
|
||||
return Response(notifications)
|
||||
|
||||
def post(self, request):
|
||||
"""Mark a single event as read."""
|
||||
event_type = request.data.get('event_type')
|
||||
event_id = request.data.get('event_id')
|
||||
if event_type not in ('deadline', 'expense') or not event_id:
|
||||
return Response({'detail': 'Invalid payload.'}, status=400)
|
||||
ReadEvent.objects.get_or_create(
|
||||
user=request.user, event_type=event_type, event_id=event_id
|
||||
)
|
||||
return Response(status=status.HTTP_204_NO_CONTENT)
|
||||
|
||||
|
||||
class ICalUrlView(views.APIView):
|
||||
"""Returns the personal iCal feed URL for the authenticated user."""
|
||||
|
||||
def get(self, request):
|
||||
token = generate_ical_token(request.user.id)
|
||||
base_url = request.build_absolute_uri('/')
|
||||
url = f"{base_url}api/calendar/ical/{request.user.id}/{token}/"
|
||||
return Response({'url': url})
|
||||
|
||||
|
||||
class ICalFeedView(views.APIView):
|
||||
"""Serves the iCal feed. Token acts as authentication — no JWT required."""
|
||||
permission_classes = [AllowAny]
|
||||
|
||||
def get(self, request, user_id, token):
|
||||
expected = generate_ical_token(user_id)
|
||||
if not hmac.compare_digest(expected, token):
|
||||
return HttpResponse(status=404)
|
||||
|
||||
User = get_user_model()
|
||||
try:
|
||||
user = User.objects.get(pk=user_id)
|
||||
except User.DoesNotExist:
|
||||
return HttpResponse(status=404)
|
||||
|
||||
cal = iCalendar()
|
||||
cal.add('prodid', '-//Budget App//EN')
|
||||
cal.add('version', '2.0')
|
||||
cal.add('x-wr-calname', 'Budget App')
|
||||
cal.add('x-wr-timezone', 'Europe/Zurich')
|
||||
|
||||
# Deadlines
|
||||
for deadline in Deadline.objects.filter(user=user):
|
||||
event = iCalEvent()
|
||||
event.add('summary', f'[{deadline.get_type_display()}] {deadline.title}')
|
||||
event.add('dtstart', deadline.date)
|
||||
event.add('dtend', deadline.date)
|
||||
event.add('uid', f'deadline-{deadline.id}@budget-app')
|
||||
if deadline.notes:
|
||||
event.add('description', deadline.notes)
|
||||
cal.add_component(event)
|
||||
|
||||
# Expense due dates
|
||||
for expense in Expense.objects.filter(account__user=user, due_date__isnull=False):
|
||||
event = iCalEvent()
|
||||
event.add('summary', f'[Invoice] {expense.name} – CHF {expense.amount}')
|
||||
event.add('dtstart', expense.due_date)
|
||||
event.add('dtend', expense.due_date)
|
||||
event.add('uid', f'expense-{expense.id}@budget-app')
|
||||
if expense.notes:
|
||||
event.add('description', expense.notes)
|
||||
cal.add_component(event)
|
||||
|
||||
response = HttpResponse(cal.to_ical(), content_type='text/calendar; charset=utf-8')
|
||||
response['Content-Disposition'] = 'attachment; filename="budget-app.ics"'
|
||||
return response
|
||||
|
||||
|
||||
# ── 2FA helpers ──────────────────────────────────────────────────────────────
|
||||
|
||||
def _make_2fa_token(user_id: int) -> str:
|
||||
"""Create a short-lived signed token binding step-1 to step-2 of login."""
|
||||
payload = f"{user_id}:{int(time.time())}"
|
||||
sig = hmac.new(settings.SECRET_KEY.encode(), payload.encode(), hashlib.sha256).hexdigest()
|
||||
return base64.urlsafe_b64encode(f"{payload}:{sig}".encode()).decode()
|
||||
|
||||
|
||||
def _verify_2fa_token(token: str, max_age: int = 300) -> int | None:
|
||||
"""Return user_id if token is valid and not expired, else None."""
|
||||
try:
|
||||
decoded = base64.urlsafe_b64decode(token.encode()).decode()
|
||||
*payload_parts, sig = decoded.split(':')
|
||||
payload = ':'.join(payload_parts)
|
||||
user_id_str, ts_str = payload_parts
|
||||
expected = hmac.new(settings.SECRET_KEY.encode(), payload.encode(), hashlib.sha256).hexdigest()
|
||||
if not hmac.compare_digest(expected, sig):
|
||||
return None
|
||||
if int(time.time()) - int(ts_str) > max_age:
|
||||
return None
|
||||
return int(user_id_str)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _generate_backup_codes(user, count: int = 8) -> list[str]:
|
||||
"""Invalidate all old backup codes and return a fresh set of plain-text codes."""
|
||||
BackupCode.objects.filter(user=user).delete()
|
||||
plain = []
|
||||
for _ in range(count):
|
||||
code = f"{secrets.token_hex(4).upper()}-{secrets.token_hex(4).upper()}"
|
||||
BackupCode.objects.create(
|
||||
user=user,
|
||||
code_hash=hashlib.sha256(code.encode()).hexdigest(),
|
||||
)
|
||||
plain.append(code)
|
||||
return plain
|
||||
|
||||
|
||||
def _verify_totp_with_replay_check(profile, code: str) -> bool:
|
||||
"""Verify TOTP code and reject replay within the same 30-second window."""
|
||||
if profile.totp_last_used_code == code:
|
||||
return False
|
||||
totp = pyotp.TOTP(profile.totp_secret)
|
||||
if not totp.verify(code, valid_window=1):
|
||||
return False
|
||||
profile.totp_last_used_code = code
|
||||
profile.save(update_fields=['totp_last_used_code'])
|
||||
return True
|
||||
|
||||
|
||||
# ── 2FA views ─────────────────────────────────────────────────────────────────
|
||||
|
||||
class LoginView(views.APIView):
|
||||
"""Replaces TokenObtainPairView. Returns a short-lived temp_token when 2FA is required."""
|
||||
permission_classes = [AllowAny]
|
||||
throttle_classes = [AuthThrottle]
|
||||
|
||||
def post(self, request):
|
||||
if not _verify_turnstile(
|
||||
request.data.get('cf_turnstile_response', ''),
|
||||
request.META.get('REMOTE_ADDR', ''),
|
||||
):
|
||||
return Response({'detail': 'Captcha verification failed.'}, status=status.HTTP_400_BAD_REQUEST)
|
||||
email = request.data.get('username', '')
|
||||
password = request.data.get('password', '')
|
||||
user = authenticate(request, username=email, password=password)
|
||||
if user is None:
|
||||
return Response({'detail': 'No active account found with the given credentials.'}, status=401)
|
||||
|
||||
profile, _ = Profile.objects.get_or_create(user=user)
|
||||
if profile.totp_enabled:
|
||||
return Response({'2fa_required': True, 'temp_token': _make_2fa_token(user.id)}, status=200)
|
||||
|
||||
refresh = RefreshToken.for_user(user)
|
||||
session_key = _create_session(user, request, refresh)
|
||||
return Response({'access': str(refresh.access_token), 'refresh': str(refresh), 'session_key': session_key})
|
||||
|
||||
|
||||
class TwoFactorLoginView(views.APIView):
|
||||
"""Step 2 of login — accepts TOTP code or backup code, returns JWT tokens."""
|
||||
permission_classes = [AllowAny]
|
||||
throttle_classes = [AuthThrottle]
|
||||
|
||||
def post(self, request):
|
||||
temp_token = request.data.get('temp_token', '')
|
||||
code = str(request.data.get('code', '')).strip()
|
||||
|
||||
user_id = _verify_2fa_token(temp_token)
|
||||
if user_id is None:
|
||||
return Response({'detail': 'Session expired. Please log in again.'}, status=401)
|
||||
|
||||
User = get_user_model()
|
||||
try:
|
||||
user = User.objects.get(pk=user_id)
|
||||
except User.DoesNotExist:
|
||||
return Response({'detail': 'Invalid credentials.'}, status=401)
|
||||
|
||||
profile, _ = Profile.objects.get_or_create(user=user)
|
||||
if not profile.totp_enabled or not profile.totp_secret:
|
||||
return Response({'detail': 'Invalid credentials.'}, status=401)
|
||||
|
||||
if code.isdigit() and len(code) == 6:
|
||||
if not _verify_totp_with_replay_check(profile, code):
|
||||
return Response({'detail': 'Invalid or already used code.'}, status=400)
|
||||
else:
|
||||
code_hash = hashlib.sha256(code.encode()).hexdigest()
|
||||
backup = BackupCode.objects.filter(user=user, code_hash=code_hash, used=False).first()
|
||||
if backup is None:
|
||||
return Response({'detail': 'Invalid backup code.'}, status=400)
|
||||
backup.used = True
|
||||
backup.save(update_fields=['used'])
|
||||
|
||||
refresh = RefreshToken.for_user(user)
|
||||
session_key = _create_session(user, request, refresh)
|
||||
return Response({'access': str(refresh.access_token), 'refresh': str(refresh), 'session_key': session_key})
|
||||
|
||||
|
||||
class TwoFactorSetupView(views.APIView):
|
||||
"""Generates a fresh TOTP secret and returns the otpauth:// URI for QR display."""
|
||||
|
||||
def get(self, request):
|
||||
profile, _ = Profile.objects.get_or_create(user=request.user)
|
||||
secret = pyotp.random_base32()
|
||||
profile.totp_secret = secret
|
||||
profile.totp_enabled = False
|
||||
profile.save(update_fields=['totp_secret', 'totp_enabled'])
|
||||
email = request.user.email or request.user.username
|
||||
uri = pyotp.TOTP(secret).provisioning_uri(name=email, issuer_name='Armarium')
|
||||
return Response({'uri': uri})
|
||||
|
||||
|
||||
class TwoFactorEnableView(views.APIView):
|
||||
"""Verifies the first TOTP code, activates 2FA and returns one-time backup codes."""
|
||||
|
||||
def post(self, request):
|
||||
code = str(request.data.get('code', '')).strip()
|
||||
profile, _ = Profile.objects.get_or_create(user=request.user)
|
||||
if not profile.totp_secret:
|
||||
return Response({'detail': 'Run setup first.'}, status=400)
|
||||
if not _verify_totp_with_replay_check(profile, code):
|
||||
return Response({'detail': 'Invalid code.'}, status=400)
|
||||
profile.totp_enabled = True
|
||||
profile.save(update_fields=['totp_enabled'])
|
||||
backup_codes = _generate_backup_codes(request.user)
|
||||
return Response({'detail': '2FA enabled.', 'backup_codes': backup_codes})
|
||||
|
||||
|
||||
class TwoFactorDisableView(views.APIView):
|
||||
"""Disables 2FA — accepts TOTP code or a backup code as proof."""
|
||||
|
||||
def post(self, request):
|
||||
code = str(request.data.get('code', '')).strip()
|
||||
profile, _ = Profile.objects.get_or_create(user=request.user)
|
||||
if not profile.totp_enabled:
|
||||
return Response({'detail': '2FA is not enabled.'}, status=400)
|
||||
|
||||
authenticated = False
|
||||
if code.isdigit() and len(code) == 6:
|
||||
authenticated = _verify_totp_with_replay_check(profile, code)
|
||||
else:
|
||||
code_hash = hashlib.sha256(code.encode()).hexdigest()
|
||||
backup = BackupCode.objects.filter(user=request.user, code_hash=code_hash, used=False).first()
|
||||
if backup:
|
||||
backup.used = True
|
||||
backup.save(update_fields=['used'])
|
||||
authenticated = True
|
||||
|
||||
if not authenticated:
|
||||
return Response({'detail': 'Invalid code.'}, status=400)
|
||||
|
||||
profile.totp_enabled = False
|
||||
profile.totp_secret = ''
|
||||
profile.totp_last_used_code = ''
|
||||
profile.save(update_fields=['totp_enabled', 'totp_secret', 'totp_last_used_code'])
|
||||
BackupCode.objects.filter(user=request.user).delete()
|
||||
return Response({'detail': '2FA disabled.'})
|
||||
|
||||
|
||||
# ── Recovery email helpers ────────────────────────────────────────────────────
|
||||
|
||||
def _mask_email(email: str) -> str:
|
||||
if '@' not in email:
|
||||
return '***'
|
||||
local, domain = email.split('@', 1)
|
||||
return f"{local[0]}{'*' * min(len(local) - 1, 18)}@{domain}"
|
||||
|
||||
|
||||
def _generate_recovery_code() -> str:
|
||||
"""Generate a human-readable 8-character code in XXXX-XXXX format."""
|
||||
alphabet = 'ABCDEFGHJKLMNPQRSTUVWXYZ23456789' # no O/0, I/1 confusion
|
||||
part = lambda: ''.join(secrets.choice(alphabet) for _ in range(4))
|
||||
return f"{part()}-{part()}"
|
||||
|
||||
|
||||
class TwoFactorRecoverRequestView(views.APIView):
|
||||
"""Generate a recovery code, store its hash in Profile and email the plain code."""
|
||||
permission_classes = [AllowAny]
|
||||
throttle_classes = [AuthThrottle]
|
||||
|
||||
def post(self, request):
|
||||
from django.utils import timezone
|
||||
from datetime import timedelta
|
||||
from .email import send_email
|
||||
|
||||
temp_token = request.data.get('temp_token', '')
|
||||
user_id = _verify_2fa_token(temp_token)
|
||||
if user_id is None:
|
||||
return Response({'detail': 'ok'})
|
||||
|
||||
User = get_user_model()
|
||||
user = User.objects.filter(pk=user_id).first()
|
||||
if not user:
|
||||
return Response({'detail': 'ok'})
|
||||
|
||||
profile = Profile.objects.filter(user=user).first()
|
||||
if not profile or not profile.recovery_email:
|
||||
return Response({'detail': 'ok'})
|
||||
|
||||
plain_code = _generate_recovery_code()
|
||||
profile.recovery_code_hash = hashlib.sha256(plain_code.encode()).hexdigest()
|
||||
profile.recovery_code_expires = timezone.now() + timedelta(minutes=15)
|
||||
profile.save(update_fields=['recovery_code_hash', 'recovery_code_expires'])
|
||||
|
||||
sent = send_email(
|
||||
template_name='2fa_recovery',
|
||||
context={'code': plain_code},
|
||||
subject='Armarium – 2FA-Wiederherstellung',
|
||||
to=profile.recovery_email,
|
||||
)
|
||||
if not sent:
|
||||
return Response({'detail': 'Failed to send recovery email.'}, status=500)
|
||||
|
||||
return Response({'detail': 'ok', 'masked_email': _mask_email(profile.recovery_email)})
|
||||
|
||||
|
||||
class TwoFactorRecoverConfirmView(views.APIView):
|
||||
"""Verify the recovery code, disable 2FA and return JWT tokens."""
|
||||
permission_classes = [AllowAny]
|
||||
throttle_classes = [AuthThrottle]
|
||||
|
||||
def post(self, request):
|
||||
from django.utils import timezone
|
||||
|
||||
temp_token = request.data.get('temp_token', '')
|
||||
user_id = _verify_2fa_token(temp_token)
|
||||
if user_id is None:
|
||||
return Response({'detail': 'Session expired. Please log in again.'}, status=401)
|
||||
|
||||
recovery_code = str(request.data.get('recovery_code', '')).strip().upper()
|
||||
if not recovery_code:
|
||||
return Response({'detail': 'Code required.'}, status=400)
|
||||
|
||||
code_hash = hashlib.sha256(recovery_code.encode()).hexdigest()
|
||||
profile = Profile.objects.filter(
|
||||
user_id=user_id,
|
||||
recovery_code_hash=code_hash,
|
||||
recovery_code_expires__gt=timezone.now(),
|
||||
).first()
|
||||
|
||||
if not profile:
|
||||
return Response({'detail': 'Invalid or expired recovery code.'}, status=400)
|
||||
|
||||
profile.totp_enabled = False
|
||||
profile.totp_secret = ''
|
||||
profile.totp_last_used_code = ''
|
||||
profile.recovery_code_hash = ''
|
||||
profile.recovery_code_expires = None
|
||||
profile.save(update_fields=[
|
||||
'totp_enabled', 'totp_secret', 'totp_last_used_code',
|
||||
'recovery_code_hash', 'recovery_code_expires',
|
||||
])
|
||||
BackupCode.objects.filter(user=profile.user).delete()
|
||||
|
||||
refresh = RefreshToken.for_user(profile.user)
|
||||
session_key = _create_session(profile.user, request, refresh)
|
||||
return Response({'access': str(refresh.access_token), 'refresh': str(refresh), 'session_key': session_key})
|
||||
|
||||
|
||||
# ── Session helpers ───────────────────────────────────────────────────────────
|
||||
|
||||
def _parse_device(ua: str) -> str:
|
||||
ua = ua.lower()
|
||||
if 'iphone' in ua: return 'iPhone'
|
||||
if 'ipad' in ua: return 'iPad'
|
||||
if 'android' in ua and 'mobile' in ua: return 'Android (Phone)'
|
||||
if 'android' in ua: return 'Android (Tablet)'
|
||||
if 'macintosh' in ua or 'mac os x' in ua: return 'Mac'
|
||||
if 'windows nt' in ua: return 'Windows'
|
||||
if 'linux' in ua: return 'Linux'
|
||||
return 'Unbekanntes Gerät'
|
||||
|
||||
|
||||
def _get_client_ip(request) -> str | None:
|
||||
forwarded = request.META.get('HTTP_X_FORWARDED_FOR', '')
|
||||
if forwarded:
|
||||
return forwarded.split(',')[0].strip()
|
||||
return request.META.get('REMOTE_ADDR') or None
|
||||
|
||||
|
||||
def _create_session(user, request, refresh_token: RefreshToken) -> str:
|
||||
session_key = secrets.token_urlsafe(32)
|
||||
UserSession.objects.create(
|
||||
user=user,
|
||||
session_key=session_key,
|
||||
refresh_jti=str(refresh_token.payload.get('jti', '')),
|
||||
device_name=_parse_device(request.META.get('HTTP_USER_AGENT', '')),
|
||||
ip_address=_get_client_ip(request),
|
||||
)
|
||||
return session_key
|
||||
|
||||
|
||||
# ── Session views ─────────────────────────────────────────────────────────────
|
||||
|
||||
class SessionListView(views.APIView):
|
||||
def get(self, request):
|
||||
current_key = request.headers.get('X-Session-Key', '')
|
||||
sessions = UserSession.objects.filter(user=request.user)
|
||||
data = [
|
||||
{
|
||||
'session_key': s.session_key,
|
||||
'device_name': s.device_name,
|
||||
'ip_address': s.ip_address,
|
||||
'created_at': s.created_at,
|
||||
'last_active_at': s.last_active_at,
|
||||
'is_current': s.session_key == current_key,
|
||||
}
|
||||
for s in sessions
|
||||
]
|
||||
return Response(data)
|
||||
|
||||
|
||||
class SessionRevokeView(views.APIView):
|
||||
def delete(self, request, session_key):
|
||||
session = UserSession.objects.filter(user=request.user, session_key=session_key).first()
|
||||
if not session:
|
||||
return Response({'detail': 'Not found.'}, status=404)
|
||||
_blacklist_session(session)
|
||||
return Response(status=204)
|
||||
|
||||
|
||||
class SessionRevokeAllView(views.APIView):
|
||||
def delete(self, request):
|
||||
current_key = request.headers.get('X-Session-Key', '')
|
||||
sessions = UserSession.objects.filter(user=request.user).exclude(session_key=current_key)
|
||||
for session in sessions:
|
||||
_blacklist_session(session)
|
||||
return Response(status=204)
|
||||
|
||||
|
||||
def _blacklist_session(session: UserSession) -> None:
|
||||
if session.refresh_jti:
|
||||
try:
|
||||
from rest_framework_simplejwt.token_blacklist.models import OutstandingToken, BlacklistedToken
|
||||
token = OutstandingToken.objects.get(jti=session.refresh_jti)
|
||||
BlacklistedToken.objects.get_or_create(token=token)
|
||||
except Exception:
|
||||
pass
|
||||
session.delete()
|
||||
|
||||
|
||||
# ── Data export ───────────────────────────────────────────────────────────────
|
||||
|
||||
class DataExportView(views.APIView):
|
||||
def get(self, request):
|
||||
import io
|
||||
import zipfile
|
||||
from datetime import date
|
||||
from fpdf import FPDF
|
||||
|
||||
user = request.user
|
||||
profile = Profile.objects.filter(user=user).first()
|
||||
today = date.today().strftime('%d.%m.%Y')
|
||||
export_date = date.today().strftime('%Y-%m-%d')
|
||||
|
||||
VIOLET = (124, 58, 237)
|
||||
HEADER_BG = (243, 240, 255)
|
||||
ALT_ROW = (249, 249, 252)
|
||||
TEXT_DARK = (30, 30, 40)
|
||||
TEXT_GRAY = (120, 120, 135)
|
||||
|
||||
def safe(text: str) -> str:
|
||||
return str(text).encode('latin-1', errors='replace').decode('latin-1')
|
||||
|
||||
class ArmPDF(FPDF):
|
||||
def __init__(self, section_title):
|
||||
super().__init__()
|
||||
self.section_title = section_title
|
||||
self.set_auto_page_break(auto=True, margin=18)
|
||||
|
||||
def header(self):
|
||||
self.set_fill_color(*VIOLET)
|
||||
self.rect(0, 0, 210, 10, 'F')
|
||||
self.set_xy(14, 13)
|
||||
self.set_font('Helvetica', 'B', 15)
|
||||
self.set_text_color(*TEXT_DARK)
|
||||
self.cell(0, 7, safe(f'Armarium - {self.section_title}'), ln=True)
|
||||
self.set_font('Helvetica', '', 8)
|
||||
self.set_text_color(*TEXT_GRAY)
|
||||
self.set_x(14)
|
||||
self.cell(0, 5, safe(f'Export vom {today}'), ln=True)
|
||||
self.ln(3)
|
||||
|
||||
def footer(self):
|
||||
self.set_y(-12)
|
||||
self.set_font('Helvetica', '', 7)
|
||||
self.set_text_color(*TEXT_GRAY)
|
||||
self.cell(0, 5, safe(f'Armarium - {today} - Seite {self.page_no()}'), align='C')
|
||||
|
||||
def table_header(self, cols):
|
||||
self.set_fill_color(*HEADER_BG)
|
||||
self.set_font('Helvetica', 'B', 8)
|
||||
self.set_text_color(*VIOLET)
|
||||
for label, width in cols:
|
||||
self.cell(width, 7, safe(label), border=0, fill=True, align='L')
|
||||
self.ln()
|
||||
self.set_draw_color(*VIOLET)
|
||||
self.set_line_width(0.4)
|
||||
x = self.get_x()
|
||||
y = self.get_y()
|
||||
self.line(14, y, 196, y)
|
||||
|
||||
def table_row(self, values, cols, fill=False):
|
||||
if fill:
|
||||
self.set_fill_color(*ALT_ROW)
|
||||
self.set_font('Helvetica', '', 8)
|
||||
self.set_text_color(*TEXT_DARK)
|
||||
for (label, width), val in zip(cols, values):
|
||||
self.cell(width, 6, safe(val), border=0, fill=fill)
|
||||
self.ln()
|
||||
|
||||
def make_pdf(title, build_fn):
|
||||
pdf = ArmPDF(title)
|
||||
pdf.add_page()
|
||||
build_fn(pdf)
|
||||
return pdf.output()
|
||||
|
||||
# ── Profil ────────────────────────────────────────────────────────────
|
||||
def build_profile(pdf):
|
||||
name = f"{profile.first_name} {profile.last_name}".strip() if profile else ''
|
||||
rows = [
|
||||
('Name', name or '-'),
|
||||
('E-Mail', user.email or '-'),
|
||||
('Kanton', profile.canton if profile else '-'),
|
||||
('Sprache', profile.language if profile else '-'),
|
||||
('2FA', 'Aktiviert' if (profile and profile.totp_enabled) else 'Deaktiviert'),
|
||||
]
|
||||
cols = [('Feld', 60), ('Wert', 120)]
|
||||
pdf.table_header(cols)
|
||||
for i, (field, val) in enumerate(rows):
|
||||
pdf.table_row([field, val], cols, fill=i % 2 == 1)
|
||||
|
||||
# ── Konten ────────────────────────────────────────────────────────────
|
||||
def build_accounts(pdf):
|
||||
cols = [('Name', 90), ('Typ', 60), ('Saldo (CHF)', 42)]
|
||||
pdf.table_header(cols)
|
||||
for i, acc in enumerate(Account.objects.filter(user=user)):
|
||||
pdf.table_row([acc.name, acc.account_type, f'{acc.balance:,.2f}'], cols, fill=i % 2 == 1)
|
||||
|
||||
# ── Budgets ───────────────────────────────────────────────────────────
|
||||
def build_budgets(pdf):
|
||||
cols = [('Name', 80), ('Kategorie', 60), ('Betrag (CHF)', 42), ('Aktiv', 10)]
|
||||
pdf.table_header(cols)
|
||||
for i, b in enumerate(Budget.objects.filter(account__user=user).order_by('main_category', 'name')):
|
||||
pdf.table_row([b.name, b.main_category, f'{b.amount:,.2f}', 'Ja' if b.active else 'Nein'], cols, fill=i % 2 == 1)
|
||||
|
||||
# ── Ausgaben ──────────────────────────────────────────────────────────
|
||||
def build_expenses(pdf):
|
||||
cols = [('Datum', 26), ('Name', 70), ('Kategorie', 46), ('Konto', 30), ('CHF', 20)]
|
||||
pdf.table_header(cols)
|
||||
for i, e in enumerate(Expense.objects.filter(account__user=user).order_by('-date')):
|
||||
pdf.table_row([
|
||||
e.date.strftime('%d.%m.%Y'), e.name, e.category,
|
||||
e.account.name, f'{e.amount:,.2f}'
|
||||
], cols, fill=i % 2 == 1)
|
||||
|
||||
# ── Transaktionen ─────────────────────────────────────────────────────
|
||||
def build_transactions(pdf):
|
||||
cols = [('Datum', 26), ('Beschreibung', 70), ('Von', 38), ('Nach', 38), ('CHF', 20)]
|
||||
pdf.table_header(cols)
|
||||
qs = Transaction.objects.filter(source_account__user=user).order_by('-date').select_related('source_account', 'destination_account')
|
||||
for i, t in enumerate(qs):
|
||||
pdf.table_row([
|
||||
t.date.strftime('%d.%m.%Y'), t.description,
|
||||
t.source_account.name, t.destination_account.name, f'{t.amount:,.2f}'
|
||||
], cols, fill=i % 2 == 1)
|
||||
|
||||
# ── Termine ───────────────────────────────────────────────────────────
|
||||
def build_deadlines(pdf):
|
||||
cols = [('Datum', 30), ('Titel', 100), ('Typ', 42), ('Notizen', 20)]
|
||||
pdf.table_header(cols)
|
||||
for i, d in enumerate(Deadline.objects.filter(user=user).order_by('date')):
|
||||
pdf.table_row([d.date.strftime('%d.%m.%Y'), d.title, d.type, d.notes[:20]], cols, fill=i % 2 == 1)
|
||||
|
||||
pdfs = [
|
||||
('profil.pdf', 'Profil', build_profile),
|
||||
('konten.pdf', 'Konten', build_accounts),
|
||||
('budgets.pdf', 'Budgets', build_budgets),
|
||||
('ausgaben.pdf', 'Ausgaben', build_expenses),
|
||||
('transaktionen.pdf', 'Transaktionen', build_transactions),
|
||||
('termine.pdf', 'Termine', build_deadlines),
|
||||
]
|
||||
|
||||
zip_buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||||
for filename, title, build_fn in pdfs:
|
||||
zf.writestr(filename, bytes(make_pdf(title, build_fn)))
|
||||
zip_buffer.seek(0)
|
||||
|
||||
response = HttpResponse(zip_buffer.read(), content_type='application/zip')
|
||||
response['Content-Disposition'] = f'attachment; filename="armarium-export-{export_date}.zip"'
|
||||
return response
|
||||
|
||||
|
||||
# ── Notification preferences ──────────────────────────────────────────────────
|
||||
|
||||
class NotificationPrefsView(views.APIView):
|
||||
def patch(self, request):
|
||||
profile, _ = Profile.objects.get_or_create(user=request.user)
|
||||
fields = ['notif_deadlines', 'notif_budget_alerts', 'notif_monthly_summary']
|
||||
changed = []
|
||||
for field in fields:
|
||||
if field in request.data:
|
||||
setattr(profile, field, bool(request.data[field]))
|
||||
changed.append(field)
|
||||
if changed:
|
||||
profile.save(update_fields=changed)
|
||||
return Response({
|
||||
'notif_deadlines': profile.notif_deadlines,
|
||||
'notif_budget_alerts': profile.notif_budget_alerts,
|
||||
'notif_monthly_summary': profile.notif_monthly_summary,
|
||||
})
|
||||
|
||||
|
||||
class VerifyEmailView(views.APIView):
|
||||
permission_classes = [AllowAny]
|
||||
throttle_classes = [AuthThrottle]
|
||||
|
||||
def post(self, request):
|
||||
from django.utils import timezone
|
||||
|
||||
token = request.data.get('token', '').strip()
|
||||
if not token:
|
||||
return Response({'detail': 'Token required.'}, status=400)
|
||||
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
||||
profile = Profile.objects.filter(
|
||||
email_verify_token=token_hash,
|
||||
email_verify_token_expires__gt=timezone.now(),
|
||||
).first()
|
||||
if not profile:
|
||||
return Response({'detail': 'Invalid or expired token.'}, status=400)
|
||||
profile.email_verified = True
|
||||
profile.email_verify_token = ''
|
||||
profile.email_verify_token_expires = None
|
||||
profile.save(update_fields=['email_verified', 'email_verify_token', 'email_verify_token_expires'])
|
||||
return Response({'detail': 'Email verified.'})
|
||||
|
||||
|
||||
class PasswordResetRequestView(views.APIView):
|
||||
permission_classes = [AllowAny]
|
||||
throttle_classes = [AuthThrottle]
|
||||
|
||||
def post(self, request):
|
||||
from django.utils import timezone
|
||||
from datetime import timedelta
|
||||
from .email import send_email
|
||||
|
||||
email = request.data.get('email', '').strip().lower()
|
||||
User = get_user_model()
|
||||
user = User.objects.filter(email=email).first()
|
||||
if user:
|
||||
token = secrets.token_urlsafe(32)
|
||||
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
||||
profile, _ = Profile.objects.get_or_create(user=user)
|
||||
profile.password_reset_token_hash = token_hash
|
||||
profile.password_reset_token_expires = timezone.now() + timedelta(minutes=15)
|
||||
profile.save(update_fields=['password_reset_token_hash', 'password_reset_token_expires'])
|
||||
link = f"{settings.FRONTEND_URL}/reset-password?token={token}"
|
||||
send_email('password_reset', {'link': link}, 'Armarium – Passwort zurücksetzen', user.email)
|
||||
return Response({'detail': 'Wenn ein Konto mit dieser E-Mail existiert, wurde eine E-Mail gesendet.'})
|
||||
|
||||
|
||||
class PasswordResetConfirmView(views.APIView):
|
||||
permission_classes = [AllowAny]
|
||||
throttle_classes = [AuthThrottle]
|
||||
|
||||
def post(self, request):
|
||||
from django.utils import timezone
|
||||
|
||||
token = request.data.get('token', '').strip()
|
||||
password = request.data.get('password', '')
|
||||
if not token:
|
||||
return Response({'detail': 'Token required.'}, status=400)
|
||||
if len(password) < 8:
|
||||
return Response({'detail': 'Password must be at least 8 characters.'}, status=400)
|
||||
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
||||
profile = Profile.objects.filter(
|
||||
password_reset_token_hash=token_hash,
|
||||
password_reset_token_expires__gt=timezone.now(),
|
||||
).first()
|
||||
if not profile:
|
||||
return Response({'detail': 'Invalid or expired token.'}, status=400)
|
||||
user = profile.user
|
||||
user.set_password(password)
|
||||
user.save()
|
||||
profile.password_reset_token_hash = ''
|
||||
profile.password_reset_token_expires = None
|
||||
profile.save(update_fields=['password_reset_token_hash', 'password_reset_token_expires'])
|
||||
for session in UserSession.objects.filter(user=user):
|
||||
_blacklist_session(session)
|
||||
return Response({'detail': 'Password updated.'})
|
||||
Reference in New Issue
Block a user