chore: rename all project references from rbox to mywebdav across configs, code, and docs

This commit is contained in:
2025-11-13 19:42:43 +00:00
parent b4a96fa82f
commit 9a7a640400
53 changed files with 74 additions and 73 deletions
View File
+17
View File
@@ -0,0 +1,17 @@
from typing import Optional
from .models import Activity, User
async def log_activity(
user: Optional[User],
action: str,
target_type: str,
target_id: int,
ip_address: Optional[str] = None
):
await Activity.create(
user=user,
action=action,
target_type=target_type,
target_id=target_id,
ip_address=ip_address
)
+83
View File
@@ -0,0 +1,83 @@
from datetime import datetime, timedelta
from typing import Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
import bcrypt
from .schemas import TokenData
from .settings import settings
from .models import User
from .two_factor import verify_totp_code # Import verify_totp_code
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def verify_password(plain_password, hashed_password):
password_bytes = plain_password[:72].encode('utf-8')
hashed_bytes = hashed_password.encode('utf-8') if isinstance(hashed_password, str) else hashed_password
return bcrypt.checkpw(password_bytes, hashed_bytes)
def get_password_hash(password):
password_bytes = password[:72].encode('utf-8')
return bcrypt.hashpw(password_bytes, bcrypt.gensalt()).decode('utf-8')
async def authenticate_user(username: str, password: str, two_factor_code: Optional[str] = None):
user = await User.get_or_none(username=username)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
if user.is_2fa_enabled:
if not two_factor_code:
return {"user": user, "2fa_required": True}
if not verify_totp_code(user.two_factor_secret, two_factor_code):
return None # 2FA code is incorrect
return {"user": user, "2fa_required": False}
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None, two_factor_verified: bool = False):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire, "2fa_verified": two_factor_verified})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
username: str = payload.get("sub")
two_factor_verified: bool = payload.get("2fa_verified", False)
if username is None:
raise credentials_exception
token_data = TokenData(username=username, two_factor_verified=two_factor_verified)
except JWTError:
raise credentials_exception
user = await User.get_or_none(username=token_data.username)
if user is None:
raise credentials_exception
user.token_data = token_data # Attach token_data to user for easy access
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)):
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
async def get_current_verified_user(current_user: User = Depends(get_current_user)):
if current_user.is_2fa_enabled and not current_user.token_data.two_factor_verified:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="2FA required and not verified")
return current_user
async def get_current_admin_user(current_user: User = Depends(get_current_verified_user)):
if not current_user.is_superuser:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
return current_user
View File
+187
View File
@@ -0,0 +1,187 @@
from datetime import datetime, date, timedelta
from decimal import Decimal
from typing import Optional
from calendar import monthrange
from .models import Invoice, InvoiceLineItem, PricingConfig, UsageAggregate, UserSubscription
from .usage_tracker import UsageTracker
from .stripe_client import StripeClient
from ..models import User
class InvoiceGenerator:
@staticmethod
async def generate_monthly_invoice(user: User, year: int, month: int) -> Optional[Invoice]:
period_start = date(year, month, 1)
days_in_month = monthrange(year, month)[1]
period_end = date(year, month, days_in_month)
usage = await UsageTracker.get_monthly_usage(user, year, month)
pricing = await PricingConfig.all()
pricing_dict = {p.config_key: p.config_value for p in pricing}
storage_price_per_gb = pricing_dict.get('storage_per_gb_month', Decimal('0.0045'))
bandwidth_price_per_gb = pricing_dict.get('bandwidth_egress_per_gb', Decimal('0.009'))
free_storage_gb = pricing_dict.get('free_tier_storage_gb', Decimal('15'))
free_bandwidth_gb = pricing_dict.get('free_tier_bandwidth_gb', Decimal('15'))
tax_rate = pricing_dict.get('tax_rate_default', Decimal('0'))
storage_gb = Decimal(str(usage['storage_gb_avg']))
bandwidth_gb = Decimal(str(usage['bandwidth_down_gb']))
billable_storage = max(Decimal('0'), storage_gb - free_storage_gb)
billable_bandwidth = max(Decimal('0'), bandwidth_gb - free_bandwidth_gb)
import math
billable_storage_rounded = Decimal(math.ceil(float(billable_storage)))
billable_bandwidth_rounded = Decimal(math.ceil(float(billable_bandwidth)))
storage_cost = billable_storage_rounded * storage_price_per_gb
bandwidth_cost = billable_bandwidth_rounded * bandwidth_price_per_gb
subtotal = storage_cost + bandwidth_cost
if subtotal <= 0:
return None
tax_amount = subtotal * tax_rate
total = subtotal + tax_amount
invoice_number = f"INV-{user.id:06d}-{year}{month:02d}"
subscription = await UserSubscription.get_or_none(user=user)
invoice = await Invoice.create(
user=user,
invoice_number=invoice_number,
period_start=period_start,
period_end=period_end,
subtotal=subtotal,
tax=tax_amount,
total=total,
currency="USD",
status="draft",
due_date=period_end + timedelta(days=7),
metadata={
"usage": usage,
"pricing": {
"storage_per_gb": float(storage_price_per_gb),
"bandwidth_per_gb": float(bandwidth_price_per_gb)
}
}
)
if billable_storage_rounded > 0:
await InvoiceLineItem.create(
invoice=invoice,
description=f"Storage usage for {period_start.strftime('%B %Y')} (Average: {storage_gb:.2f} GB, Billable: {billable_storage_rounded} GB)",
quantity=billable_storage_rounded,
unit_price=storage_price_per_gb,
amount=storage_cost,
item_type="storage",
metadata={"avg_gb": float(storage_gb), "free_gb": float(free_storage_gb)}
)
if billable_bandwidth_rounded > 0:
await InvoiceLineItem.create(
invoice=invoice,
description=f"Bandwidth usage for {period_start.strftime('%B %Y')} (Total: {bandwidth_gb:.2f} GB, Billable: {billable_bandwidth_rounded} GB)",
quantity=billable_bandwidth_rounded,
unit_price=bandwidth_price_per_gb,
amount=bandwidth_cost,
item_type="bandwidth",
metadata={"total_gb": float(bandwidth_gb), "free_gb": float(free_bandwidth_gb)}
)
if subscription and subscription.stripe_customer_id:
try:
line_items = await invoice.line_items.all()
stripe_line_items = [
{
"amount": item.amount,
"currency": "usd",
"description": item.description,
"metadata": item.metadata or {}
}
for item in line_items
]
stripe_invoice = await StripeClient.create_invoice(
customer_id=subscription.stripe_customer_id,
description=f"MyWebdav Usage Invoice for {period_start.strftime('%B %Y')}",
line_items=stripe_line_items,
metadata={"mywebdav_invoice_id": str(invoice.id)}
)
invoice.stripe_invoice_id = stripe_invoice.id
await invoice.save()
except Exception as e:
print(f"Failed to create Stripe invoice: {e}")
return invoice
@staticmethod
async def finalize_invoice(invoice: Invoice) -> Invoice:
if invoice.status != "draft":
raise ValueError("Only draft invoices can be finalized")
invoice.status = "open"
await invoice.save()
if invoice.stripe_invoice_id:
try:
await StripeClient.finalize_invoice(invoice.stripe_invoice_id)
except Exception as e:
print(f"Failed to finalize Stripe invoice: {e}")
# Send invoice email
from ..mail import queue_email
line_items = await invoice.line_items.all()
items_text = "\n".join([f"- {item.description}: ${item.amount}" for item in line_items])
body = f"""Dear {invoice.user.username},
Your invoice {invoice.invoice_number} for the period {invoice.period_start} to {invoice.period_end} is now available.
Invoice Details:
{items_text}
Subtotal: ${invoice.subtotal}
Tax: ${invoice.tax}
Total: ${invoice.total}
Due Date: {invoice.due_date}
You can view and pay your invoice at: {invoice.user.email} # Placeholder, should be a link to invoice page
Best regards,
The MyWebdav Team
"""
html = f"""
<h2>Invoice {invoice.invoice_number}</h2>
<p>Dear {invoice.user.username},</p>
<p>Your invoice for the period {invoice.period_start} to {invoice.period_end} is now available.</p>
<table border="1">
<tr><th>Description</th><th>Amount</th></tr>
{"".join([f"<tr><td>{item.description}</td><td>${item.amount}</td></tr>" for item in line_items])}
<tr><td><strong>Subtotal</strong></td><td><strong>${invoice.subtotal}</strong></td></tr>
<tr><td><strong>Tax</strong></td><td><strong>${invoice.tax}</strong></td></tr>
<tr><td><strong>Total</strong></td><td><strong>${invoice.total}</strong></td></tr>
</table>
<p>Due Date: {invoice.due_date}</p>
<p>You can view and pay your invoice at: <a href="#">Invoice Link</a></p>
<p>Best regards,<br>The MyWebdav Team</p>
"""
queue_email(
to_email=invoice.user.email,
subject=f"Your RBox Invoice {invoice.invoice_number}",
body=body,
html=html
)
return invoice
@staticmethod
async def mark_invoice_paid(invoice: Invoice) -> Invoice:
invoice.status = "paid"
invoice.paid_at = datetime.utcnow()
await invoice.save()
return invoice
+141
View File
@@ -0,0 +1,141 @@
from tortoise import fields, models
from decimal import Decimal
class SubscriptionPlan(models.Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=100, unique=True)
display_name = fields.CharField(max_length=255)
description = fields.TextField(null=True)
storage_gb = fields.IntField()
bandwidth_gb = fields.IntField()
price_monthly = fields.DecimalField(max_digits=10, decimal_places=2)
price_yearly = fields.DecimalField(max_digits=10, decimal_places=2, null=True)
stripe_price_id = fields.CharField(max_length=255, null=True)
is_active = fields.BooleanField(default=True)
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)
class Meta:
table = "subscription_plans"
class UserSubscription(models.Model):
id = fields.IntField(pk=True)
user = fields.ForeignKeyField("models.User", related_name="subscription")
plan = fields.ForeignKeyField("billing.SubscriptionPlan", related_name="subscriptions", null=True)
billing_type = fields.CharField(max_length=20, default="pay_as_you_go")
stripe_customer_id = fields.CharField(max_length=255, unique=True, null=True)
stripe_subscription_id = fields.CharField(max_length=255, unique=True, null=True)
status = fields.CharField(max_length=50, default="active")
current_period_start = fields.DatetimeField(null=True)
current_period_end = fields.DatetimeField(null=True)
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)
canceled_at = fields.DatetimeField(null=True)
class Meta:
table = "user_subscriptions"
class UsageRecord(models.Model):
id = fields.BigIntField(pk=True)
user = fields.ForeignKeyField("models.User", related_name="usage_records")
record_type = fields.CharField(max_length=50, index=True)
amount_bytes = fields.BigIntField()
resource_type = fields.CharField(max_length=50, null=True)
resource_id = fields.IntField(null=True)
timestamp = fields.DatetimeField(auto_now_add=True, index=True)
idempotency_key = fields.CharField(max_length=255, unique=True, null=True)
metadata = fields.JSONField(null=True)
class Meta:
table = "usage_records"
indexes = [("user_id", "record_type", "timestamp")]
class UsageAggregate(models.Model):
id = fields.IntField(pk=True)
user = fields.ForeignKeyField("models.User", related_name="usage_aggregates")
date = fields.DateField()
storage_bytes_avg = fields.BigIntField(default=0)
storage_bytes_peak = fields.BigIntField(default=0)
bandwidth_up_bytes = fields.BigIntField(default=0)
bandwidth_down_bytes = fields.BigIntField(default=0)
created_at = fields.DatetimeField(auto_now_add=True)
class Meta:
table = "usage_aggregates"
unique_together = (("user", "date"),)
class Invoice(models.Model):
id = fields.IntField(pk=True)
user = fields.ForeignKeyField("models.User", related_name="invoices")
invoice_number = fields.CharField(max_length=50, unique=True)
stripe_invoice_id = fields.CharField(max_length=255, unique=True, null=True)
period_start = fields.DateField(index=True)
period_end = fields.DateField()
subtotal = fields.DecimalField(max_digits=10, decimal_places=4)
tax = fields.DecimalField(max_digits=10, decimal_places=4, default=0)
total = fields.DecimalField(max_digits=10, decimal_places=4)
currency = fields.CharField(max_length=3, default="USD")
status = fields.CharField(max_length=50, default="draft", index=True)
due_date = fields.DateField(null=True)
paid_at = fields.DatetimeField(null=True)
created_at = fields.DatetimeField(auto_now_add=True, index=True)
updated_at = fields.DatetimeField(auto_now=True)
metadata = fields.JSONField(null=True)
class Meta:
table = "invoices"
indexes = [("user_id", "status", "created_at")]
class InvoiceLineItem(models.Model):
id = fields.IntField(pk=True)
invoice = fields.ForeignKeyField("billing.Invoice", related_name="line_items")
description = fields.TextField()
quantity = fields.DecimalField(max_digits=15, decimal_places=6)
unit_price = fields.DecimalField(max_digits=10, decimal_places=6)
amount = fields.DecimalField(max_digits=10, decimal_places=4)
item_type = fields.CharField(max_length=50, null=True)
metadata = fields.JSONField(null=True)
created_at = fields.DatetimeField(auto_now_add=True)
class Meta:
table = "invoice_line_items"
class PricingConfig(models.Model):
id = fields.IntField(pk=True)
config_key = fields.CharField(max_length=100, unique=True)
config_value = fields.DecimalField(max_digits=10, decimal_places=6)
description = fields.TextField(null=True)
unit = fields.CharField(max_length=50, null=True)
updated_by = fields.ForeignKeyField("models.User", related_name="pricing_updates", null=True)
updated_at = fields.DatetimeField(auto_now=True)
class Meta:
table = "pricing_config"
class PaymentMethod(models.Model):
id = fields.IntField(pk=True)
user = fields.ForeignKeyField("models.User", related_name="payment_methods")
stripe_payment_method_id = fields.CharField(max_length=255)
type = fields.CharField(max_length=50)
is_default = fields.BooleanField(default=False)
last4 = fields.CharField(max_length=4, null=True)
brand = fields.CharField(max_length=50, null=True)
exp_month = fields.IntField(null=True)
exp_year = fields.IntField(null=True)
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)
class Meta:
table = "payment_methods"
class BillingEvent(models.Model):
id = fields.BigIntField(pk=True)
user = fields.ForeignKeyField("models.User", related_name="billing_events", null=True)
event_type = fields.CharField(max_length=100)
stripe_event_id = fields.CharField(max_length=255, unique=True, null=True)
data = fields.JSONField(null=True)
processed = fields.BooleanField(default=False)
created_at = fields.DatetimeField(auto_now_add=True)
class Meta:
table = "billing_events"
+57
View File
@@ -0,0 +1,57 @@
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime, date, timedelta
import asyncio
from .usage_tracker import UsageTracker
from .invoice_generator import InvoiceGenerator
from ..models import User
scheduler = AsyncIOScheduler()
async def aggregate_daily_usage_for_all_users():
users = await User.filter(is_active=True).all()
yesterday = date.today() - timedelta(days=1)
for user in users:
try:
await UsageTracker.aggregate_daily_usage(user, yesterday)
except Exception as e:
print(f"Failed to aggregate usage for user {user.id}: {e}")
async def generate_monthly_invoices():
now = datetime.now()
last_month = now.month - 1 if now.month > 1 else 12
year = now.year if now.month > 1 else now.year - 1
users = await User.filter(is_active=True).all()
for user in users:
try:
invoice = await InvoiceGenerator.generate_monthly_invoice(user, year, last_month)
if invoice:
await InvoiceGenerator.finalize_invoice(invoice)
except Exception as e:
print(f"Failed to generate invoice for user {user.id}: {e}")
def start_scheduler():
scheduler.add_job(
aggregate_daily_usage_for_all_users,
CronTrigger(hour=1, minute=0),
id="aggregate_daily_usage",
name="Aggregate daily usage for all users",
replace_existing=True
)
scheduler.add_job(
generate_monthly_invoices,
CronTrigger(day=1, hour=2, minute=0),
id="generate_monthly_invoices",
name="Generate monthly invoices",
replace_existing=True
)
scheduler.start()
def stop_scheduler():
scheduler.shutdown()
+119
View File
@@ -0,0 +1,119 @@
import stripe
from decimal import Decimal
from typing import Optional, Dict, Any
from ..settings import settings
class StripeClient:
@staticmethod
def _ensure_api_key():
if not stripe.api_key:
if settings.STRIPE_SECRET_KEY:
stripe.api_key = settings.STRIPE_SECRET_KEY
else:
raise ValueError("Stripe API key not configured")
@staticmethod
async def create_customer(email: str, name: str, metadata: Dict = None) -> str:
StripeClient._ensure_api_key()
customer = stripe.Customer.create(
email=email,
name=name,
metadata=metadata or {}
)
return customer.id
@staticmethod
async def create_payment_intent(
amount: int,
currency: str = "usd",
customer_id: str = None,
metadata: Dict = None
) -> stripe.PaymentIntent:
StripeClient._ensure_api_key()
return stripe.PaymentIntent.create(
amount=amount,
currency=currency,
customer=customer_id,
metadata=metadata or {},
automatic_payment_methods={"enabled": True}
)
@staticmethod
async def create_invoice(
customer_id: str,
description: str,
line_items: list,
metadata: Dict = None
) -> stripe.Invoice:
StripeClient._ensure_api_key()
for item in line_items:
stripe.InvoiceItem.create(
customer=customer_id,
amount=int(item['amount'] * 100),
currency=item.get('currency', 'usd'),
description=item['description'],
metadata=item.get('metadata', {})
)
invoice = stripe.Invoice.create(
customer=customer_id,
description=description,
auto_advance=True,
collection_method='charge_automatically',
metadata=metadata or {}
)
return invoice
@staticmethod
async def finalize_invoice(invoice_id: str) -> stripe.Invoice:
StripeClient._ensure_api_key()
return stripe.Invoice.finalize_invoice(invoice_id)
@staticmethod
async def pay_invoice(invoice_id: str) -> stripe.Invoice:
StripeClient._ensure_api_key()
return stripe.Invoice.pay(invoice_id)
@staticmethod
async def attach_payment_method(
payment_method_id: str,
customer_id: str
) -> stripe.PaymentMethod:
StripeClient._ensure_api_key()
payment_method = stripe.PaymentMethod.attach(
payment_method_id,
customer=customer_id
)
stripe.Customer.modify(
customer_id,
invoice_settings={'default_payment_method': payment_method_id}
)
return payment_method
@staticmethod
async def list_payment_methods(customer_id: str, type: str = "card"):
StripeClient._ensure_api_key()
return stripe.PaymentMethod.list(
customer=customer_id,
type=type
)
@staticmethod
async def create_subscription(
customer_id: str,
price_id: str,
metadata: Dict = None
) -> stripe.Subscription:
StripeClient._ensure_api_key()
return stripe.Subscription.create(
customer=customer_id,
items=[{'price': price_id}],
metadata=metadata or {}
)
@staticmethod
async def cancel_subscription(subscription_id: str) -> stripe.Subscription:
StripeClient._ensure_api_key()
return stripe.Subscription.delete(subscription_id)
+149
View File
@@ -0,0 +1,149 @@
import uuid
from datetime import datetime, date
from decimal import Decimal
from typing import Optional
from tortoise.transactions import in_transaction
from .models import UsageRecord, UsageAggregate
from ..models import User
class UsageTracker:
@staticmethod
async def track_storage(
user: User,
amount_bytes: int,
resource_type: str = None,
resource_id: int = None,
metadata: dict = None
):
idempotency_key = f"storage_{user.id}_{datetime.utcnow().timestamp()}_{uuid.uuid4().hex[:8]}"
await UsageRecord.create(
user=user,
record_type="storage",
amount_bytes=amount_bytes,
resource_type=resource_type,
resource_id=resource_id,
idempotency_key=idempotency_key,
metadata=metadata
)
@staticmethod
async def track_bandwidth(
user: User,
amount_bytes: int,
direction: str = "down",
resource_type: str = None,
resource_id: int = None,
metadata: dict = None
):
record_type = f"bandwidth_{direction}"
idempotency_key = f"{record_type}_{user.id}_{datetime.utcnow().timestamp()}_{uuid.uuid4().hex[:8]}"
await UsageRecord.create(
user=user,
record_type=record_type,
amount_bytes=amount_bytes,
resource_type=resource_type,
resource_id=resource_id,
idempotency_key=idempotency_key,
metadata=metadata
)
@staticmethod
async def aggregate_daily_usage(user: User, target_date: date = None):
if target_date is None:
target_date = date.today()
start_of_day = datetime.combine(target_date, datetime.min.time())
end_of_day = datetime.combine(target_date, datetime.max.time())
storage_records = await UsageRecord.filter(
user=user,
record_type="storage",
timestamp__gte=start_of_day,
timestamp__lte=end_of_day
).all()
storage_avg = sum(r.amount_bytes for r in storage_records) // max(len(storage_records), 1)
storage_peak = max((r.amount_bytes for r in storage_records), default=0)
bandwidth_up = await UsageRecord.filter(
user=user,
record_type="bandwidth_up",
timestamp__gte=start_of_day,
timestamp__lte=end_of_day
).all()
bandwidth_down = await UsageRecord.filter(
user=user,
record_type="bandwidth_down",
timestamp__gte=start_of_day,
timestamp__lte=end_of_day
).all()
total_up = sum(r.amount_bytes for r in bandwidth_up)
total_down = sum(r.amount_bytes for r in bandwidth_down)
async with in_transaction():
aggregate, created = await UsageAggregate.get_or_create(
user=user,
date=target_date,
defaults={
"storage_bytes_avg": storage_avg,
"storage_bytes_peak": storage_peak,
"bandwidth_up_bytes": total_up,
"bandwidth_down_bytes": total_down
}
)
if not created:
aggregate.storage_bytes_avg = storage_avg
aggregate.storage_bytes_peak = storage_peak
aggregate.bandwidth_up_bytes = total_up
aggregate.bandwidth_down_bytes = total_down
await aggregate.save()
return aggregate
@staticmethod
async def get_current_storage(user: User) -> int:
from ..models import File
files = await File.filter(owner=user, is_deleted=False).all()
return sum(f.size for f in files)
@staticmethod
async def get_monthly_usage(user: User, year: int, month: int) -> dict:
from datetime import date
from calendar import monthrange
start_date = date(year, month, 1)
_, last_day = monthrange(year, month)
end_date = date(year, month, last_day)
aggregates = await UsageAggregate.filter(
user=user,
date__gte=start_date,
date__lte=end_date
).all()
if not aggregates:
return {
"storage_gb_avg": 0,
"storage_gb_peak": 0,
"bandwidth_up_gb": 0,
"bandwidth_down_gb": 0,
"total_bandwidth_gb": 0
}
storage_avg = sum(a.storage_bytes_avg for a in aggregates) / len(aggregates)
storage_peak = max(a.storage_bytes_peak for a in aggregates)
bandwidth_up = sum(a.bandwidth_up_bytes for a in aggregates)
bandwidth_down = sum(a.bandwidth_down_bytes for a in aggregates)
return {
"storage_gb_avg": round(storage_avg / (1024**3), 4),
"storage_gb_peak": round(storage_peak / (1024**3), 4),
"bandwidth_up_gb": round(bandwidth_up / (1024**3), 4),
"bandwidth_down_gb": round(bandwidth_down / (1024**3), 4),
"total_bandwidth_gb": round((bandwidth_up + bandwidth_down) / (1024**3), 4)
}
+124
View File
@@ -0,0 +1,124 @@
import asyncio
import aiosmtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Optional, Dict, Any
from .settings import settings
class EmailTask:
def __init__(self, to_email: str, subject: str, body: str, html: Optional[str] = None, **kwargs):
self.to_email = to_email
self.subject = subject
self.body = body
self.html = html
self.kwargs = kwargs
class EmailService:
def __init__(self):
self.queue = asyncio.Queue()
self.worker_task: Optional[asyncio.Task] = None
self.running = False
async def start(self):
"""Start the email worker"""
if self.running:
return
self.running = True
self.worker_task = asyncio.create_task(self._worker())
async def stop(self):
"""Stop the email worker"""
if not self.running:
return
self.running = False
if self.worker_task:
self.worker_task.cancel()
try:
await self.worker_task
except asyncio.CancelledError:
pass
async def send_email(self, to_email: str, subject: str, body: str, html: Optional[str] = None, **kwargs):
"""Queue an email for sending"""
task = EmailTask(to_email, subject, body, html, **kwargs)
await self.queue.put(task)
async def _worker(self):
"""Email worker coroutine"""
while self.running:
try:
# Wait for a task with timeout to allow checking running flag
task = await asyncio.wait_for(self.queue.get(), timeout=1.0)
await self._send_email_task(task)
self.queue.task_done()
except asyncio.TimeoutError:
continue
except Exception as e:
# Log error, but continue processing
print(f"Email worker error: {e}")
continue
async def _send_email_task(self, task: EmailTask):
"""Send a single email task"""
if not settings.SMTP_HOST or not settings.SMTP_USERNAME or not settings.SMTP_PASSWORD:
print("SMTP not configured, skipping email send")
return
msg = MIMEMultipart('alternative')
msg['From'] = settings.SMTP_SENDER_EMAIL
msg['To'] = task.to_email
msg['Subject'] = task.subject
# Add text part
text_part = MIMEText(task.body, 'plain')
msg.attach(text_part)
# Add HTML part if provided
if task.html:
html_part = MIMEText(task.html, 'html')
msg.attach(html_part)
try:
# Use implicit TLS for port 465 or when SMTP_USE_TLS is True
if settings.SMTP_PORT == 465 or settings.SMTP_USE_TLS:
async with aiosmtplib.SMTP(
hostname=settings.SMTP_HOST,
port=settings.SMTP_PORT,
username=settings.SMTP_USERNAME,
password=settings.SMTP_PASSWORD,
use_tls=True
) as smtp:
await smtp.send_message(msg)
print(f"Email sent to {task.to_email}")
else:
# Use STARTTLS for other ports
async with aiosmtplib.SMTP(
hostname=settings.SMTP_HOST,
port=settings.SMTP_PORT,
) as smtp:
try:
await smtp.starttls()
except Exception as tls_error:
if "already using" in str(tls_error).lower() or "tls" in str(tls_error).lower():
# Connection is already using TLS, proceed without starttls
pass
else:
raise
await smtp.login(settings.SMTP_USERNAME, settings.SMTP_PASSWORD)
await smtp.send_message(msg)
print(f"Email sent to {task.to_email}")
except Exception as e:
print(f"Failed to send email to {task.to_email}: {e}")
raise # Re-raise to let caller handle
# Global email service instance
email_service = EmailService()
# Convenience functions
async def send_email(to_email: str, subject: str, body: str, html: Optional[str] = None, **kwargs):
"""Send an email asynchronously"""
await email_service.send_email(to_email, subject, body, html, **kwargs)
def queue_email(to_email: str, subject: str, body: str, html: Optional[str] = None, **kwargs):
"""Queue an email for sending (fire and forget)"""
asyncio.create_task(email_service.send_email(to_email, subject, body, html, **kwargs))
+107
View File
@@ -0,0 +1,107 @@
import argparse
import uvicorn
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, status, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse, JSONResponse
from tortoise.contrib.fastapi import register_tortoise
from .settings import settings
from .routers import auth, users, folders, files, shares, search, admin, starred, billing, admin_billing
from . import webdav
from .schemas import ErrorResponse
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("Starting up...")
logger.info("Database connected.")
from .billing.scheduler import start_scheduler
from .billing.models import PricingConfig
from .mail import email_service
start_scheduler()
logger.info("Billing scheduler started")
await email_service.start()
logger.info("Email service started")
pricing_count = await PricingConfig.all().count()
if pricing_count == 0:
from decimal import Decimal
await PricingConfig.create(config_key='storage_per_gb_month', config_value=Decimal('0.0045'), description='Storage cost per GB per month', unit='per_gb_month')
await PricingConfig.create(config_key='bandwidth_egress_per_gb', config_value=Decimal('0.009'), description='Bandwidth egress cost per GB', unit='per_gb')
await PricingConfig.create(config_key='bandwidth_ingress_per_gb', config_value=Decimal('0.0'), description='Bandwidth ingress cost per GB (free)', unit='per_gb')
await PricingConfig.create(config_key='free_tier_storage_gb', config_value=Decimal('15'), description='Free tier storage in GB', unit='gb')
await PricingConfig.create(config_key='free_tier_bandwidth_gb', config_value=Decimal('15'), description='Free tier bandwidth in GB per month', unit='gb')
await PricingConfig.create(config_key='tax_rate_default', config_value=Decimal('0.0'), description='Default tax rate (0 = no tax)', unit='percentage')
logger.info("Default pricing configuration initialized")
yield
from .billing.scheduler import stop_scheduler
stop_scheduler()
logger.info("Billing scheduler stopped")
await email_service.stop()
logger.info("Email service stopped")
print("Shutting down...")
app = FastAPI(
title="MyWebdav Cloud Storage",
description="A commercial cloud storage web application",
version="0.1.0",
lifespan=lifespan
)
app.include_router(auth.router)
app.include_router(users.router)
app.include_router(folders.router)
app.include_router(files.router)
app.include_router(shares.router)
app.include_router(search.router)
app.include_router(admin.router)
app.include_router(starred.router)
app.include_router(billing.router)
app.include_router(admin_billing.router)
app.include_router(webdav.router)
from .middleware.usage_tracking import UsageTrackingMiddleware
app.add_middleware(UsageTrackingMiddleware)
app.mount("/static", StaticFiles(directory="static"), name="static")
register_tortoise(
app,
db_url=settings.DATABASE_URL,
modules={
"models": ["mywebdav.models"],
"billing": ["mywebdav.billing.models"]
},
generate_schemas=True,
add_exception_handlers=True,
)
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
logger.error(f"HTTPException: {exc.status_code} - {exc.detail} for URL: {request.url}")
return JSONResponse(
status_code=exc.status_code,
content=ErrorResponse(code=exc.status_code, message=exc.detail).dict(),
)
@app.get("/", response_class=HTMLResponse) # Change response_class to HTMLResponse
async def read_root():
with open("static/index.html", "r") as f:
return f.read()
def main():
parser = argparse.ArgumentParser(description="Run the RBox application.")
parser.add_argument("--host", type=str, default="0.0.0.0", help="Host address to bind to")
parser.add_argument("--port", type=int, default=8000, help="Port to listen on")
args = parser.parse_args()
uvicorn.run(app, host=args.host, port=args.port)
if __name__ == "__main__":
main()
View File
+32
View File
@@ -0,0 +1,32 @@
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from ..billing.usage_tracker import UsageTracker
class UsageTrackingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
response = await call_next(request)
if hasattr(request.state, 'user') and request.state.user:
user = request.state.user
if request.method in ['POST', 'PUT'] and '/files/upload' in request.url.path:
content_length = response.headers.get('content-length')
if content_length:
await UsageTracker.track_bandwidth(
user=user,
amount_bytes=int(content_length),
direction='up',
metadata={'path': request.url.path}
)
elif request.method == 'GET' and '/files/download' in request.url.path:
content_length = response.headers.get('content-length')
if content_length:
await UsageTracker.track_bandwidth(
user=user,
amount_bytes=int(content_length),
direction='down',
metadata={'path': request.url.path}
)
return response
+155
View File
@@ -0,0 +1,155 @@
from tortoise import fields, models
from tortoise.contrib.pydantic import pydantic_model_creator
from datetime import datetime
class User(models.Model):
id = fields.IntField(pk=True)
username = fields.CharField(max_length=20, unique=True)
email = fields.CharField(max_length=255, unique=True)
hashed_password = fields.CharField(max_length=255)
is_active = fields.BooleanField(default=True)
is_superuser = fields.BooleanField(default=False)
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)
storage_quota_bytes = fields.BigIntField(default=10 * 1024 * 1024 * 1024) # 10 GB default
used_storage_bytes = fields.BigIntField(default=0)
plan_type = fields.CharField(max_length=50, default="free")
two_factor_secret = fields.CharField(max_length=255, null=True)
is_2fa_enabled = fields.BooleanField(default=False)
recovery_codes = fields.TextField(null=True)
class Meta:
table = "users"
def __str__(self):
return self.username
class Folder(models.Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=255)
parent: fields.ForeignKeyRelation["Folder"] = fields.ForeignKeyField("models.Folder", related_name="children", null=True)
owner: fields.ForeignKeyRelation[User] = fields.ForeignKeyField("models.User", related_name="folders")
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)
is_deleted = fields.BooleanField(default=False)
is_starred = fields.BooleanField(default=False)
class Meta:
table = "folders"
unique_together = (("name", "parent", "owner"),) # Ensure unique folder names within a parent for an owner
def __str__(self):
return self.name
class File(models.Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=255)
path = fields.CharField(max_length=1024) # Internal storage path
size = fields.BigIntField()
mime_type = fields.CharField(max_length=255)
file_hash = fields.CharField(max_length=64, null=True) # SHA-256
thumbnail_path = fields.CharField(max_length=1024, null=True)
parent: fields.ForeignKeyRelation[Folder] = fields.ForeignKeyField("models.Folder", related_name="files", null=True)
owner: fields.ForeignKeyRelation[User] = fields.ForeignKeyField("models.User", related_name="files")
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)
is_deleted = fields.BooleanField(default=False)
deleted_at = fields.DatetimeField(null=True)
is_starred = fields.BooleanField(default=False)
last_accessed_at = fields.DatetimeField(null=True)
class Meta:
table = "files"
unique_together = (("name", "parent", "owner"),) # Ensure unique file names within a parent for an owner
def __str__(self):
return self.name
class FileVersion(models.Model):
id = fields.IntField(pk=True)
file: fields.ForeignKeyRelation[File] = fields.ForeignKeyField("models.File", related_name="versions")
version_path = fields.CharField(max_length=1024)
size = fields.BigIntField()
created_at = fields.DatetimeField(auto_now_add=True)
class Meta:
table = "file_versions"
class Share(models.Model):
id = fields.IntField(pk=True)
token = fields.CharField(max_length=64, unique=True)
file: fields.ForeignKeyRelation[File] = fields.ForeignKeyField("models.File", related_name="shares", null=True)
folder: fields.ForeignKeyRelation[Folder] = fields.ForeignKeyField("models.Folder", related_name="shares", null=True)
owner: fields.ForeignKeyRelation[User] = fields.ForeignKeyField("models.User", related_name="shares")
created_at = fields.DatetimeField(auto_now_add=True)
expires_at = fields.DatetimeField(null=True)
password_protected = fields.BooleanField(default=False)
hashed_password = fields.CharField(max_length=255, null=True)
access_count = fields.IntField(default=0)
permission_level = fields.CharField(max_length=50, default="viewer") # viewer, uploader, editor
class Meta:
table = "shares"
class Team(models.Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=255, unique=True)
owner: fields.ForeignKeyRelation[User] = fields.ForeignKeyField("models.User", related_name="owned_teams")
members: fields.ManyToManyRelation[User] = fields.ManyToManyField("models.User", related_name="teams", through="team_members")
created_at = fields.DatetimeField(auto_now_add=True)
class Meta:
table = "teams"
class TeamMember(models.Model):
id = fields.IntField(pk=True)
team: fields.ForeignKeyRelation[Team] = fields.ForeignKeyField("models.Team", related_name="team_members")
user: fields.ForeignKeyRelation[User] = fields.ForeignKeyField("models.User", related_name="user_teams")
role = fields.CharField(max_length=50, default="member") # owner, admin, member
class Meta:
table = "team_members"
unique_together = (("team", "user"),)
class Activity(models.Model):
id = fields.IntField(pk=True)
user: fields.ForeignKeyRelation[User] = fields.ForeignKeyField("models.User", related_name="activities", null=True)
action = fields.CharField(max_length=255)
target_type = fields.CharField(max_length=50) # file, folder, share, user, team
target_id = fields.IntField()
ip_address = fields.CharField(max_length=45, null=True)
timestamp = fields.DatetimeField(auto_now_add=True)
class Meta:
table = "activities"
class FileRequest(models.Model):
id = fields.IntField(pk=True)
title = fields.CharField(max_length=255)
description = fields.TextField(null=True)
token = fields.CharField(max_length=64, unique=True)
owner: fields.ForeignKeyRelation[User] = fields.ForeignKeyField("models.User", related_name="file_requests")
target_folder: fields.ForeignKeyRelation[Folder] = fields.ForeignKeyField("models.Folder", related_name="file_requests")
created_at = fields.DatetimeField(auto_now_add=True)
expires_at = fields.DatetimeField(null=True)
is_active = fields.BooleanField(default=True)
class Meta:
table = "file_requests"
class WebDAVProperty(models.Model):
id = fields.IntField(pk=True)
resource_type = fields.CharField(max_length=10)
resource_id = fields.IntField()
namespace = fields.CharField(max_length=255)
name = fields.CharField(max_length=255)
value = fields.TextField()
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)
class Meta:
table = "webdav_properties"
unique_together = (("resource_type", "resource_id", "namespace", "name"),)
User_Pydantic = pydantic_model_creator(User, name="User_Pydantic")
UserIn_Pydantic = pydantic_model_creator(User, name="UserIn_Pydantic", exclude_readonly=True)
+109
View File
@@ -0,0 +1,109 @@
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, status
from ..auth import get_current_admin_user, get_password_hash
from ..models import User, User_Pydantic
from ..schemas import UserCreate, UserAdminUpdate
router = APIRouter(
prefix="/admin",
tags=["admin"],
dependencies=[Depends(get_current_admin_user)],
responses={403: {"description": "Not enough permissions"}},
)
@router.get("/users", response_model=List[User_Pydantic])
async def get_all_users():
return await User.all()
@router.get("/users/{user_id}", response_model=User_Pydantic)
async def get_user(user_id: int):
user = await User.get_or_none(id=user_id)
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
return user
@router.post("/users", response_model=User_Pydantic, status_code=status.HTTP_201_CREATED)
async def create_user_by_admin(user_in: UserCreate):
user = await User.get_or_none(username=user_in.username)
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already registered",
)
user = await User.get_or_none(email=user_in.email)
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered",
)
hashed_password = get_password_hash(user_in.password)
user = await User.create(
username=user_in.username,
email=user_in.email,
hashed_password=hashed_password,
is_superuser=False, # Admin creates regular users by default
is_active=True,
)
return await User_Pydantic.from_tortoise_orm(user)
@router.put("/users/{user_id}", response_model=User_Pydantic)
async def update_user_by_admin(user_id: int, user_update: UserAdminUpdate):
user = await User.get_or_none(id=user_id)
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
if user_update.username is not None and user_update.username != user.username:
if await User.get_or_none(username=user_update.username):
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Username already taken")
user.username = user_update.username
if user_update.email is not None and user_update.email != user.email:
if await User.get_or_none(email=user_update.email):
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered")
user.email = user_update.email
if user_update.password is not None:
user.hashed_password = get_password_hash(user_update.password)
if user_update.is_active is not None:
user.is_active = user_update.is_active
if user_update.is_superuser is not None:
user.is_superuser = user_update.is_superuser
if user_update.storage_quota_bytes is not None:
user.storage_quota_bytes = user_update.storage_quota_bytes
if user_update.plan_type is not None:
user.plan_type = user_update.plan_type
if user_update.is_2fa_enabled is not None:
user.is_2fa_enabled = user_update.is_2fa_enabled
if not user_update.is_2fa_enabled:
user.two_factor_secret = None
user.recovery_codes = None
await user.save()
return await User_Pydantic.from_tortoise_orm(user)
@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user_by_admin(user_id: int):
user = await User.get_or_none(id=user_id)
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
await user.delete()
return {"message": "User deleted successfully"}
@router.post("/test-email")
async def send_test_email(to_email: str, subject: str = "Test Email", body: str = "This is a test email"):
from ..mail import queue_email
queue_email(
to_email=to_email,
subject=subject,
body=body,
html=f"<h1>{subject}</h1><p>{body}</p>"
)
return {"message": "Test email queued"}
+116
View File
@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException
from typing import List
from decimal import Decimal
from pydantic import BaseModel
from ..auth import get_current_user
from ..models import User
from ..billing.models import PricingConfig, Invoice, SubscriptionPlan
from ..billing.invoice_generator import InvoiceGenerator
def require_superuser(current_user: User = Depends(get_current_user)):
if not current_user.is_superuser:
raise HTTPException(status_code=403, detail="Superuser privileges required")
return current_user
router = APIRouter(
prefix="/api/admin/billing",
tags=["admin", "billing"],
dependencies=[Depends(require_superuser)]
)
class PricingConfigUpdate(BaseModel):
config_key: str
config_value: float
class PlanCreate(BaseModel):
name: str
display_name: str
description: str
storage_gb: int
bandwidth_gb: int
price_monthly: float
price_yearly: float = None
@router.get("/pricing")
async def get_all_pricing(current_user: User = Depends(require_superuser)):
configs = await PricingConfig.all()
return [
{
"id": c.id,
"config_key": c.config_key,
"config_value": float(c.config_value),
"description": c.description,
"unit": c.unit,
"updated_at": c.updated_at
}
for c in configs
]
@router.put("/pricing/{config_id}")
async def update_pricing(
config_id: int,
update: PricingConfigUpdate,
current_user: User = Depends(require_superuser)
):
config = await PricingConfig.get_or_none(id=config_id)
if not config:
raise HTTPException(status_code=404, detail="Config not found")
config.config_value = Decimal(str(update.config_value))
config.updated_by = current_user
await config.save()
return {"message": "Pricing updated successfully"}
@router.post("/generate-invoices/{year}/{month}")
async def generate_all_invoices(
year: int,
month: int,
current_user: User = Depends(require_superuser)
):
users = await User.filter(is_active=True).all()
generated = []
skipped = []
for user in users:
invoice = await InvoiceGenerator.generate_monthly_invoice(user, year, month)
if invoice:
generated.append({
"user_id": user.id,
"invoice_id": invoice.id,
"total": float(invoice.total)
})
else:
skipped.append(user.id)
return {
"generated": len(generated),
"skipped": len(skipped),
"invoices": generated
}
@router.post("/plans")
async def create_plan(
plan_data: PlanCreate,
current_user: User = Depends(require_superuser)
):
plan = await SubscriptionPlan.create(**plan_data.dict())
return {"id": plan.id, "message": "Plan created successfully"}
@router.get("/stats")
async def get_billing_stats(current_user: User = Depends(require_superuser)):
from tortoise.functions import Sum, Count
total_revenue = await Invoice.filter(status="paid").annotate(
total_sum=Sum("total")
).values("total_sum")
invoice_count = await Invoice.all().count()
pending_invoices = await Invoice.filter(status="open").count()
return {
"total_revenue": float(total_revenue[0]["total_sum"] or 0),
"total_invoices": invoice_count,
"pending_invoices": pending_invoices
}
+174
View File
@@ -0,0 +1,174 @@
from datetime import timedelta
from typing import Optional, List
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel
from ..auth import authenticate_user, create_access_token, get_password_hash, get_current_user, get_current_verified_user, verify_password
from ..models import User
from ..schemas import Token, UserCreate, TokenData, UserLoginWith2FA
from ..two_factor import (
generate_totp_secret, generate_totp_uri, generate_qr_code_base64,
verify_totp_code, generate_recovery_codes, hash_recovery_codes,
verify_recovery_codes
)
router = APIRouter(
prefix="/auth",
tags=["auth"],
)
class LoginRequest(BaseModel):
username: str
password: str
class TwoFactorLogin(BaseModel):
username: str
password: str
two_factor_code: Optional[str] = None
class TwoFactorSetupResponse(BaseModel):
secret: str
qr_code_base64: str
recovery_codes: List[str]
class TwoFactorCode(BaseModel):
two_factor_code: str
class TwoFactorDisable(BaseModel):
password: str
two_factor_code: str
@router.post("/register", response_model=Token)
async def register_user(user_in: UserCreate):
user = await User.get_or_none(username=user_in.username)
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already registered",
)
user = await User.get_or_none(email=user_in.email)
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered",
)
hashed_password = get_password_hash(user_in.password)
user = await User.create(
username=user_in.username,
email=user_in.email,
hashed_password=hashed_password,
)
# Send welcome email
from ..mail import queue_email
queue_email(
to_email=user.email,
subject="Welcome to MyWebdav!",
body=f"Hi {user.username},\n\nWelcome to MyWebdav! Your account has been created successfully.\n\nBest regards,\nThe MyWebdav Team",
html=f"<h1>Welcome to MyWebdav!</h1><p>Hi {user.username},</p><p>Welcome to MyWebdav! Your account has been created successfully.</p><p>Best regards,<br>The MyWebdav Team</p>"
)
access_token_expires = timedelta(minutes=30) # Use settings
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@router.post("/token", response_model=Token)
async def login_for_access_token(login_data: LoginRequest):
auth_result = await authenticate_user(login_data.username, login_data.password, None)
if not auth_result:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
user = auth_result["user"]
if auth_result["2fa_required"]:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Two-factor authentication required",
headers={"X-2FA-Required": "true"},
)
access_token_expires = timedelta(minutes=30)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires, two_factor_verified=True
)
return {"access_token": access_token, "token_type": "bearer"}
@router.post("/2fa/setup", response_model=TwoFactorSetupResponse)
async def setup_two_factor_authentication(current_user: User = Depends(get_current_user)):
if current_user.is_2fa_enabled:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="2FA is already enabled.")
if current_user.two_factor_secret:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="2FA setup already initiated. Verify or disable first.")
secret = generate_totp_secret()
current_user.two_factor_secret = secret
await current_user.save()
totp_uri = generate_totp_uri(secret, current_user.email, "RBox")
qr_code_base64 = generate_qr_code_base64(totp_uri)
recovery_codes = generate_recovery_codes()
hashed_recovery_codes = hash_recovery_codes(recovery_codes)
current_user.recovery_codes = ",".join(hashed_recovery_codes)
await current_user.save()
return TwoFactorSetupResponse(secret=secret, qr_code_base64=qr_code_base64, recovery_codes=recovery_codes)
@router.post("/2fa/verify", response_model=Token)
async def verify_two_factor_authentication(two_factor_code_data: TwoFactorCode, current_user: User = Depends(get_current_user)):
if current_user.is_2fa_enabled:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="2FA is already enabled.")
if not current_user.two_factor_secret:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="2FA setup not initiated.")
if not verify_totp_code(current_user.two_factor_secret, two_factor_code_data.two_factor_code):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid 2FA code.")
current_user.is_2fa_enabled = True
await current_user.save()
access_token_expires = timedelta(minutes=30) # Use settings
access_token = create_access_token(
data={"sub": current_user.username}, expires_delta=access_token_expires, two_factor_verified=True
)
return {"access_token": access_token, "token_type": "bearer"}
@router.post("/2fa/disable", response_model=dict)
async def disable_two_factor_authentication(disable_data: TwoFactorDisable, current_user: User = Depends(get_current_verified_user)):
if not current_user.is_2fa_enabled:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="2FA is not enabled.")
# Verify password
if not verify_password(disable_data.password, current_user.hashed_password):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid password.")
# Verify 2FA code
if not verify_totp_code(current_user.two_factor_secret, disable_data.two_factor_code):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid 2FA code.")
current_user.two_factor_secret = None
current_user.is_2fa_enabled = False
current_user.recovery_codes = None
await current_user.save()
return {"message": "2FA disabled successfully."}
@router.get("/2fa/recovery-codes", response_model=List[str])
async def get_new_recovery_codes(current_user: User = Depends(get_current_verified_user)):
if not current_user.is_2fa_enabled:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="2FA is not enabled.")
recovery_codes = generate_recovery_codes()
hashed_recovery_codes = hash_recovery_codes(recovery_codes)
current_user.recovery_codes = ",".join(hashed_recovery_codes)
await current_user.save()
return recovery_codes
+374
View File
@@ -0,0 +1,374 @@
from fastapi import APIRouter, Depends, HTTPException, status, Request
from fastapi.responses import JSONResponse
from typing import List, Optional
from datetime import datetime, date
from decimal import Decimal
import calendar
from ..auth import get_current_user
from ..models import User
from ..billing.models import (
Invoice, InvoiceLineItem, UserSubscription, PricingConfig,
PaymentMethod, UsageAggregate, SubscriptionPlan
)
from ..billing.usage_tracker import UsageTracker
from ..billing.invoice_generator import InvoiceGenerator
from ..billing.stripe_client import StripeClient
from pydantic import BaseModel
router = APIRouter(
prefix="/api/billing",
tags=["billing"]
)
class UsageResponse(BaseModel):
storage_gb_avg: float
storage_gb_peak: float
bandwidth_up_gb: float
bandwidth_down_gb: float
total_bandwidth_gb: float
period: str
class InvoiceResponse(BaseModel):
id: int
invoice_number: str
period_start: date
period_end: date
subtotal: float
tax: float
total: float
status: str
due_date: Optional[date]
paid_at: Optional[datetime]
line_items: List[dict]
class SubscriptionResponse(BaseModel):
id: int
billing_type: str
plan_name: Optional[str]
status: str
current_period_start: Optional[datetime]
current_period_end: Optional[datetime]
@router.get("/usage/current")
async def get_current_usage(current_user: User = Depends(get_current_user)):
try:
storage_bytes = await UsageTracker.get_current_storage(current_user)
today = date.today()
usage_today = await UsageAggregate.get_or_none(user=current_user, date=today)
if usage_today:
return {
"storage_gb": round(storage_bytes / (1024**3), 4),
"bandwidth_down_gb_today": round(usage_today.bandwidth_down_bytes / (1024**3), 4),
"bandwidth_up_gb_today": round(usage_today.bandwidth_up_bytes / (1024**3), 4),
"as_of": today.isoformat()
}
return {
"storage_gb": round(storage_bytes / (1024**3), 4),
"bandwidth_down_gb_today": 0,
"bandwidth_up_gb_today": 0,
"as_of": today.isoformat()
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to fetch usage data: {str(e)}")
@router.get("/usage/monthly")
async def get_monthly_usage(
year: Optional[int] = None,
month: Optional[int] = None,
current_user: User = Depends(get_current_user)
) -> UsageResponse:
try:
if year is None or month is None:
now = datetime.now()
year = now.year
month = now.month
if not (1 <= month <= 12):
raise HTTPException(status_code=400, detail="Month must be between 1 and 12")
if not (2020 <= year <= 2100):
raise HTTPException(status_code=400, detail="Year must be between 2020 and 2100")
usage = await UsageTracker.get_monthly_usage(current_user, year, month)
return UsageResponse(
**usage,
period=f"{year}-{month:02d}"
)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to fetch monthly usage: {str(e)}")
@router.get("/invoices")
async def list_invoices(
limit: int = 50,
offset: int = 0,
current_user: User = Depends(get_current_user)
) -> List[InvoiceResponse]:
try:
if limit < 1 or limit > 100:
raise HTTPException(status_code=400, detail="Limit must be between 1 and 100")
if offset < 0:
raise HTTPException(status_code=400, detail="Offset must be non-negative")
invoices = await Invoice.filter(user=current_user).order_by("-created_at").offset(offset).limit(limit).all()
result = []
for invoice in invoices:
line_items = await invoice.line_items.all()
result.append(InvoiceResponse(
id=invoice.id,
invoice_number=invoice.invoice_number,
period_start=invoice.period_start,
period_end=invoice.period_end,
subtotal=float(invoice.subtotal),
tax=float(invoice.tax),
total=float(invoice.total),
status=invoice.status,
due_date=invoice.due_date,
paid_at=invoice.paid_at,
line_items=[
{
"description": item.description,
"quantity": float(item.quantity),
"unit_price": float(item.unit_price),
"amount": float(item.amount),
"type": item.item_type
}
for item in line_items
]
))
return result
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to fetch invoices: {str(e)}")
@router.get("/invoices/{invoice_id}")
async def get_invoice(
invoice_id: int,
current_user: User = Depends(get_current_user)
) -> InvoiceResponse:
invoice = await Invoice.get_or_none(id=invoice_id, user=current_user)
if not invoice:
raise HTTPException(status_code=404, detail="Invoice not found")
line_items = await invoice.line_items.all()
return InvoiceResponse(
id=invoice.id,
invoice_number=invoice.invoice_number,
period_start=invoice.period_start,
period_end=invoice.period_end,
subtotal=float(invoice.subtotal),
tax=float(invoice.tax),
total=float(invoice.total),
status=invoice.status,
due_date=invoice.due_date,
paid_at=invoice.paid_at,
line_items=[
{
"description": item.description,
"quantity": float(item.quantity),
"unit_price": float(item.unit_price),
"amount": float(item.amount),
"type": item.item_type
}
for item in line_items
]
)
@router.get("/subscription")
async def get_subscription(current_user: User = Depends(get_current_user)) -> SubscriptionResponse:
subscription = await UserSubscription.get_or_none(user=current_user)
if not subscription:
subscription = await UserSubscription.create(
user=current_user,
billing_type="pay_as_you_go",
status="active"
)
plan_name = None
if subscription.plan:
plan = await subscription.plan
plan_name = plan.display_name
return SubscriptionResponse(
id=subscription.id,
billing_type=subscription.billing_type,
plan_name=plan_name,
status=subscription.status,
current_period_start=subscription.current_period_start,
current_period_end=subscription.current_period_end
)
@router.post("/payment-methods/setup-intent")
async def create_setup_intent(current_user: User = Depends(get_current_user)):
try:
from ..settings import settings
if not settings.STRIPE_SECRET_KEY:
raise HTTPException(status_code=503, detail="Payment processing not configured")
subscription = await UserSubscription.get_or_none(user=current_user)
if not subscription or not subscription.stripe_customer_id:
customer_id = await StripeClient.create_customer(
email=current_user.email,
name=current_user.username,
metadata={"user_id": str(current_user.id)}
)
if not subscription:
subscription = await UserSubscription.create(
user=current_user,
billing_type="pay_as_you_go",
stripe_customer_id=customer_id,
status="active"
)
else:
subscription.stripe_customer_id = customer_id
await subscription.save()
import stripe
StripeClient._ensure_api_key()
setup_intent = stripe.SetupIntent.create(
customer=subscription.stripe_customer_id,
payment_method_types=["card"]
)
return {
"client_secret": setup_intent.client_secret,
"customer_id": subscription.stripe_customer_id
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to create setup intent: {str(e)}")
@router.get("/payment-methods")
async def list_payment_methods(current_user: User = Depends(get_current_user)):
methods = await PaymentMethod.filter(user=current_user).all()
return [
{
"id": m.id,
"type": m.type,
"last4": m.last4,
"brand": m.brand,
"exp_month": m.exp_month,
"exp_year": m.exp_year,
"is_default": m.is_default
}
for m in methods
]
@router.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
import stripe
from ..settings import settings
from ..billing.models import BillingEvent
try:
payload = await request.body()
sig_header = request.headers.get("stripe-signature")
if not settings.STRIPE_WEBHOOK_SECRET:
raise HTTPException(status_code=503, detail="Webhook secret not configured")
try:
event = stripe.Webhook.construct_event(
payload, sig_header, settings.STRIPE_WEBHOOK_SECRET
)
except ValueError as e:
raise HTTPException(status_code=400, detail=f"Invalid payload: {str(e)}")
except stripe.error.SignatureVerificationError as e:
raise HTTPException(status_code=400, detail=f"Invalid signature: {str(e)}")
event_id = event.get("id")
existing_event = await BillingEvent.get_or_none(stripe_event_id=event_id)
if existing_event:
return JSONResponse(content={"status": "already_processed"})
await BillingEvent.create(
event_type=event["type"],
stripe_event_id=event_id,
data=event["data"],
processed=False
)
if event["type"] == "invoice.payment_succeeded":
invoice_data = event["data"]["object"]
mywebdav_invoice_id = invoice_data.get("metadata", {}).get("mywebdav_invoice_id")
if mywebdav_invoice_id:
invoice = await Invoice.get_or_none(id=int(mywebdav_invoice_id))
if invoice:
await InvoiceGenerator.mark_invoice_paid(invoice)
elif event["type"] == "invoice.payment_failed":
pass
elif event["type"] == "payment_method.attached":
payment_method = event["data"]["object"]
customer_id = payment_method["customer"]
subscription = await UserSubscription.get_or_none(stripe_customer_id=customer_id)
if subscription:
await PaymentMethod.create(
user=subscription.user,
stripe_payment_method_id=payment_method["id"],
type=payment_method["type"],
last4=payment_method.get("card", {}).get("last4"),
brand=payment_method.get("card", {}).get("brand"),
exp_month=payment_method.get("card", {}).get("exp_month"),
exp_year=payment_method.get("card", {}).get("exp_year"),
is_default=True
)
await BillingEvent.filter(stripe_event_id=event_id).update(processed=True)
return JSONResponse(content={"status": "success"})
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Webhook processing failed: {str(e)}")
@router.get("/pricing")
async def get_pricing():
configs = await PricingConfig.all()
return {
config.config_key: {
"value": float(config.config_value),
"description": config.description,
"unit": config.unit
}
for config in configs
}
@router.get("/plans")
async def list_plans():
plans = await SubscriptionPlan.filter(is_active=True).all()
return [
{
"id": plan.id,
"name": plan.name,
"display_name": plan.display_name,
"description": plan.description,
"storage_gb": plan.storage_gb,
"bandwidth_gb": plan.bandwidth_gb,
"price_monthly": float(plan.price_monthly),
"price_yearly": float(plan.price_yearly) if plan.price_yearly else None
}
for plan in plans
]
@router.get("/stripe-key")
async def get_stripe_key():
from ..settings import settings
if not settings.STRIPE_PUBLISHABLE_KEY:
raise HTTPException(status_code=503, detail="Payment processing not configured")
return {"publishable_key": settings.STRIPE_PUBLISHABLE_KEY}
+482
View File
@@ -0,0 +1,482 @@
from fastapi import APIRouter, Depends, UploadFile, File as FastAPIFile, HTTPException, status, Response, Form
from fastapi.responses import StreamingResponse
from typing import List, Optional
import mimetypes
import hashlib
import os
from datetime import datetime
from pydantic import BaseModel
from ..auth import get_current_user
from ..models import User, File, Folder
from ..schemas import FileOut
from ..storage import storage_manager
from ..settings import settings
from ..activity import log_activity
from ..thumbnails import generate_thumbnail, delete_thumbnail
router = APIRouter(
prefix="/files",
tags=["files"],
)
class FileMove(BaseModel):
target_folder_id: Optional[int] = None
class FileRename(BaseModel):
new_name: str
class FileCopy(BaseModel):
target_folder_id: Optional[int] = None
class BatchFileOperation(BaseModel):
file_ids: List[int]
operation: str # e.g., "delete", "star", "unstar", "move", "copy"
class BatchMoveCopyPayload(BaseModel):
target_folder_id: Optional[int] = None
class FileContentUpdate(BaseModel):
content: str
@router.post("/upload", response_model=FileOut, status_code=status.HTTP_201_CREATED)
async def upload_file(
file: UploadFile = FastAPIFile(...),
folder_id: Optional[int] = Form(None),
current_user: User = Depends(get_current_user)
):
if folder_id:
parent_folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not parent_folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
else:
parent_folder = None
existing_file = await File.get_or_none(
name=file.filename, parent=parent_folder, owner=current_user, is_deleted=False
)
if existing_file:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="File with this name already exists in the current folder",
)
file_content = await file.read()
file_size = len(file_content)
file_hash = hashlib.sha256(file_content).hexdigest()
if current_user.used_storage_bytes + file_size > current_user.storage_quota_bytes:
raise HTTPException(
status_code=status.HTTP_507_INSUFFICIENT_STORAGE,
detail="Storage quota exceeded",
)
# Generate a unique path for storage
file_extension = os.path.splitext(file.filename)[1]
unique_filename = f"{file_hash}{file_extension}" # Use hash for unique filename
storage_path = unique_filename
# Save file to storage
await storage_manager.save_file(current_user.id, storage_path, file_content)
# Get mime type
mime_type, _ = mimetypes.guess_type(file.filename)
if not mime_type:
mime_type = "application/octet-stream"
# Create file entry in database
db_file = await File.create(
name=file.filename,
path=storage_path,
size=file_size,
mime_type=mime_type,
file_hash=file_hash,
owner=current_user,
parent=parent_folder,
)
current_user.used_storage_bytes += file_size
await current_user.save()
thumbnail_path = await generate_thumbnail(storage_path, mime_type, current_user.id)
if thumbnail_path:
db_file.thumbnail_path = thumbnail_path
await db_file.save()
return await FileOut.from_tortoise_orm(db_file)
@router.get("/download/{file_id}")
async def download_file(file_id: int, current_user: User = Depends(get_current_user)):
db_file = await File.get_or_none(id=file_id, owner=current_user, is_deleted=False)
if not db_file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
db_file.last_accessed_at = datetime.now()
await db_file.save()
try:
async def file_iterator():
async for chunk in storage_manager.get_file(current_user.id, db_file.path):
yield chunk
return StreamingResponse(
file_iterator(),
media_type=db_file.mime_type,
headers={"Content-Disposition": f"attachment; filename=\"{db_file.name}\""}
)
except FileNotFoundError:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found in storage")
@router.delete("/{file_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_file(file_id: int, current_user: User = Depends(get_current_user)):
db_file = await File.get_or_none(id=file_id, owner=current_user, is_deleted=False)
if not db_file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
db_file.is_deleted = True
db_file.deleted_at = datetime.now()
await db_file.save()
await delete_thumbnail(db_file.id)
return
@router.post("/{file_id}/move", response_model=FileOut)
async def move_file(file_id: int, move_data: FileMove, current_user: User = Depends(get_current_user)):
db_file = await File.get_or_none(id=file_id, owner=current_user, is_deleted=False)
if not db_file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
target_folder = None
if move_data.target_folder_id:
target_folder = await Folder.get_or_none(id=move_data.target_folder_id, owner=current_user, is_deleted=False)
if not target_folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Target folder not found")
existing_file = await File.get_or_none(
name=db_file.name, parent=target_folder, owner=current_user, is_deleted=False
)
if existing_file and existing_file.id != file_id:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="File with this name already exists in target folder")
db_file.parent = target_folder
await db_file.save()
await log_activity(user=current_user, action="file_moved", target_type="file", target_id=file_id)
return await FileOut.from_tortoise_orm(db_file)
@router.post("/{file_id}/rename", response_model=FileOut)
async def rename_file(file_id: int, rename_data: FileRename, current_user: User = Depends(get_current_user)):
db_file = await File.get_or_none(id=file_id, owner=current_user, is_deleted=False)
if not db_file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
existing_file = await File.get_or_none(
name=rename_data.new_name, parent_id=db_file.parent_id, owner=current_user, is_deleted=False
)
if existing_file and existing_file.id != file_id:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="File with this name already exists in the same folder")
db_file.name = rename_data.new_name
await db_file.save()
await log_activity(user=current_user, action="file_renamed", target_type="file", target_id=file_id)
return await FileOut.from_tortoise_orm(db_file)
@router.post("/{file_id}/copy", response_model=FileOut, status_code=status.HTTP_201_CREATED)
async def copy_file(file_id: int, copy_data: FileCopy, current_user: User = Depends(get_current_user)):
db_file = await File.get_or_none(id=file_id, owner=current_user, is_deleted=False)
if not db_file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
target_folder = None
if copy_data.target_folder_id:
target_folder = await Folder.get_or_none(id=copy_data.target_folder_id, owner=current_user, is_deleted=False)
if not target_folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Target folder not found")
base_name = db_file.name
name_parts = os.path.splitext(base_name)
counter = 1
new_name = base_name
while await File.get_or_none(name=new_name, parent=target_folder, owner=current_user, is_deleted=False):
new_name = f"{name_parts[0]} (copy {counter}){name_parts[1]}"
counter += 1
new_file = await File.create(
name=new_name,
path=db_file.path,
size=db_file.size,
mime_type=db_file.mime_type,
file_hash=db_file.file_hash,
owner=current_user,
parent=target_folder
)
await log_activity(user=current_user, action="file_copied", target_type="file", target_id=new_file.id)
return await FileOut.from_tortoise_orm(new_file)
@router.get("/", response_model=List[FileOut])
async def list_files(folder_id: Optional[int] = None, current_user: User = Depends(get_current_user)):
if folder_id:
parent_folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not parent_folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
files = await File.filter(parent=parent_folder, owner=current_user, is_deleted=False).order_by("name")
else:
files = await File.filter(parent=None, owner=current_user, is_deleted=False).order_by("name")
return [await FileOut.from_tortoise_orm(f) for f in files]
@router.get("/thumbnail/{file_id}")
async def get_thumbnail(file_id: int, current_user: User = Depends(get_current_user)):
db_file = await File.get_or_none(id=file_id, owner=current_user, is_deleted=False)
if not db_file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
db_file.last_accessed_at = datetime.now()
await db_file.save()
thumbnail_path = getattr(db_file, 'thumbnail_path', None)
if not thumbnail_path:
thumbnail_path = await generate_thumbnail(db_file.path, db_file.mime_type, current_user.id)
if thumbnail_path:
db_file.thumbnail_path = thumbnail_path
await db_file.save()
else:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Thumbnail not available")
try:
async def thumbnail_iterator():
async for chunk in storage_manager.get_file(current_user.id, thumbnail_path):
yield chunk
return StreamingResponse(
thumbnail_iterator(),
media_type="image/jpeg"
)
except FileNotFoundError:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Thumbnail not found in storage")
@router.get("/photos", response_model=List[FileOut])
async def list_photos(current_user: User = Depends(get_current_user)):
files = await File.filter(
owner=current_user,
is_deleted=False,
mime_type__istartswith="image/"
).order_by("-created_at")
return [await FileOut.from_tortoise_orm(f) for f in files]
@router.get("/recent", response_model=List[FileOut])
async def list_recent_files(current_user: User = Depends(get_current_user), limit: int = 10):
files = await File.filter(
owner=current_user,
is_deleted=False,
last_accessed_at__isnull=False
).order_by("-last_accessed_at").limit(limit)
return [await FileOut.from_tortoise_orm(f) for f in files]
@router.post("/{file_id}/star", response_model=FileOut)
async def star_file(file_id: int, current_user: User = Depends(get_current_user)):
db_file = await File.get_or_none(id=file_id, owner=current_user, is_deleted=False)
if not db_file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
db_file.is_starred = True
await db_file.save()
await log_activity(user=current_user, action="file_starred", target_type="file", target_id=file_id)
return await FileOut.from_tortoise_orm(db_file)
@router.post("/{file_id}/unstar", response_model=FileOut)
async def unstar_file(file_id: int, current_user: User = Depends(get_current_user)):
db_file = await File.get_or_none(id=file_id, owner=current_user, is_deleted=False)
if not db_file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
db_file.is_starred = False
await db_file.save()
await log_activity(user=current_user, action="file_unstarred", target_type="file", target_id=file_id)
return await FileOut.from_tortoise_orm(db_file)
@router.get("/deleted", response_model=List[FileOut])
async def list_deleted_files(current_user: User = Depends(get_current_user)):
files = await File.filter(owner=current_user, is_deleted=True).order_by("-deleted_at")
return [await FileOut.from_tortoise_orm(f) for f in files]
@router.post("/{file_id}/restore", response_model=FileOut)
async def restore_file(file_id: int, current_user: User = Depends(get_current_user)):
db_file = await File.get_or_none(id=file_id, owner=current_user, is_deleted=True)
if not db_file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Deleted file not found")
# Check if a file with the same name exists in the parent folder
existing_file = await File.get_or_none(
name=db_file.name, parent=db_file.parent, owner=current_user, is_deleted=False
)
if existing_file:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="A file with the same name already exists in this location. Please rename the existing file or restore to a different location.")
db_file.is_deleted = False
db_file.deleted_at = None
await db_file.save()
await log_activity(user=current_user, action="file_restored", target_type="file", target_id=file_id)
return await FileOut.from_tortoise_orm(db_file)
class BatchOperationResult(BaseModel):
succeeded: List[FileOut]
failed: List[dict]
@router.post("/batch")
async def batch_file_operations(
batch_operation: BatchFileOperation,
payload: Optional[BatchMoveCopyPayload] = None,
current_user: User = Depends(get_current_user)
):
if batch_operation.operation not in ["delete", "star", "unstar", "move", "copy"]:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid operation: {batch_operation.operation}")
updated_files = []
failed_operations = []
for file_id in batch_operation.file_ids:
try:
db_file = await File.get_or_none(id=file_id, owner=current_user, is_deleted=False)
if not db_file:
failed_operations.append({"file_id": file_id, "reason": "File not found or not owned by user"})
continue
if batch_operation.operation == "delete":
db_file.is_deleted = True
db_file.deleted_at = datetime.now()
await db_file.save()
await delete_thumbnail(db_file.id)
await log_activity(user=current_user, action="file_deleted_batch", target_type="file", target_id=file_id)
updated_files.append(db_file)
elif batch_operation.operation == "star":
db_file.is_starred = True
await db_file.save()
await log_activity(user=current_user, action="file_starred_batch", target_type="file", target_id=file_id)
updated_files.append(db_file)
elif batch_operation.operation == "unstar":
db_file.is_starred = False
await db_file.save()
await log_activity(user=current_user, action="file_unstarred_batch", target_type="file", target_id=file_id)
updated_files.append(db_file)
elif batch_operation.operation == "move":
if not payload or payload.target_folder_id is None:
failed_operations.append({"file_id": file_id, "reason": "Target folder not specified"})
continue
target_folder = await Folder.get_or_none(id=payload.target_folder_id, owner=current_user, is_deleted=False)
if not target_folder:
failed_operations.append({"file_id": file_id, "reason": "Target folder not found"})
continue
existing_file = await File.get_or_none(
name=db_file.name, parent=target_folder, owner=current_user, is_deleted=False
)
if existing_file and existing_file.id != file_id:
failed_operations.append({"file_id": file_id, "reason": "File with same name exists in target folder"})
continue
db_file.parent = target_folder
await db_file.save()
await log_activity(user=current_user, action="file_moved_batch", target_type="file", target_id=file_id)
updated_files.append(db_file)
elif batch_operation.operation == "copy":
if not payload or payload.target_folder_id is None:
failed_operations.append({"file_id": file_id, "reason": "Target folder not specified"})
continue
target_folder = await Folder.get_or_none(id=payload.target_folder_id, owner=current_user, is_deleted=False)
if not target_folder:
failed_operations.append({"file_id": file_id, "reason": "Target folder not found"})
continue
base_name = db_file.name
name_parts = os.path.splitext(base_name)
counter = 1
new_name = base_name
while await File.get_or_none(name=new_name, parent=target_folder, owner=current_user, is_deleted=False):
new_name = f"{name_parts[0]} (copy {counter}){name_parts[1]}"
counter += 1
new_file = await File.create(
name=new_name,
path=db_file.path,
size=db_file.size,
mime_type=db_file.mime_type,
file_hash=db_file.file_hash,
owner=current_user,
parent=target_folder
)
await log_activity(user=current_user, action="file_copied_batch", target_type="file", target_id=new_file.id)
updated_files.append(new_file)
except Exception as e:
failed_operations.append({"file_id": file_id, "reason": str(e)})
return {
"succeeded": [await FileOut.from_tortoise_orm(f) for f in updated_files],
"failed": failed_operations
}
@router.put("/{file_id}/content", response_model=FileOut)
async def update_file_content(
file_id: int,
payload: FileContentUpdate,
current_user: User = Depends(get_current_user)
):
db_file = await File.get_or_none(id=file_id, owner=current_user, is_deleted=False)
if not db_file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
if not db_file.mime_type or not db_file.mime_type.startswith('text/'):
editableExtensions = [
'txt', 'md', 'log', 'json', 'js', 'py', 'html', 'css',
'xml', 'yaml', 'yml', 'sh', 'bat', 'ini', 'conf', 'cfg'
]
file_extension = os.path.splitext(db_file.name)[1][1:].lower()
if file_extension not in editableExtensions:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="File type is not editable"
)
content_bytes = payload.content.encode('utf-8')
new_size = len(content_bytes)
size_diff = new_size - db_file.size
if current_user.used_storage_bytes + size_diff > current_user.storage_quota_bytes:
raise HTTPException(
status_code=status.HTTP_507_INSUFFICIENT_STORAGE,
detail="Storage quota exceeded"
)
new_hash = hashlib.sha256(content_bytes).hexdigest()
file_extension = os.path.splitext(db_file.name)[1]
new_storage_path = f"{new_hash}{file_extension}"
await storage_manager.save_file(current_user.id, new_storage_path, content_bytes)
if new_storage_path != db_file.path:
try:
await storage_manager.delete_file(current_user.id, db_file.path)
except:
pass
db_file.path = new_storage_path
db_file.size = new_size
db_file.file_hash = new_hash
db_file.updated_at = datetime.utcnow()
await db_file.save()
current_user.used_storage_bytes += size_diff
await current_user.save()
await log_activity(user=current_user, action="file_updated", target_type="file", target_id=file_id)
return await FileOut.from_tortoise_orm(db_file)
+190
View File
@@ -0,0 +1,190 @@
from fastapi import APIRouter, Depends, HTTPException, status
from typing import List, Optional
from ..auth import get_current_user
from ..models import User, Folder
from ..schemas import FolderCreate, FolderOut, FolderUpdate, BatchFolderOperation, BatchMoveCopyPayload
from ..activity import log_activity
router = APIRouter(
prefix="/folders",
tags=["folders"],
)
@router.post("/", response_model=FolderOut, status_code=status.HTTP_201_CREATED)
async def create_folder(folder_in: FolderCreate, current_user: User = Depends(get_current_user)):
# Check if parent folder exists and belongs to the current user
parent_folder = None
if folder_in.parent_id:
parent_folder = await Folder.get_or_none(id=folder_in.parent_id, owner=current_user)
if not parent_folder:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Parent folder not found or does not belong to the current user",
)
# Check for duplicate folder name in the same parent
existing_folder = await Folder.get_or_none(
name=folder_in.name, parent=parent_folder, owner=current_user
)
if existing_folder:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Folder with this name already exists in the current parent folder",
)
folder = await Folder.create(
name=folder_in.name, parent=parent_folder, owner=current_user
)
await log_activity(current_user, "folder_created", "folder", folder.id)
return await FolderOut.from_tortoise_orm(folder)
@router.get("/{folder_id}/path", response_model=List[FolderOut])
async def get_folder_path(folder_id: int, current_user: User = Depends(get_current_user)):
folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
path = []
current = folder
while current:
path.insert(0, await FolderOut.from_tortoise_orm(current))
if current.parent_id:
current = await Folder.get_or_none(id=current.parent_id, owner=current_user, is_deleted=False)
else:
current = None
return path
@router.get("/{folder_id}", response_model=FolderOut)
async def get_folder(folder_id: int, current_user: User = Depends(get_current_user)):
folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
return await FolderOut.from_tortoise_orm(folder)
@router.get("/", response_model=List[FolderOut])
async def list_folders(parent_id: Optional[int] = None, current_user: User = Depends(get_current_user)):
if parent_id:
parent_folder = await Folder.get_or_none(id=parent_id, owner=current_user, is_deleted=False)
if not parent_folder:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Parent folder not found or does not belong to the current user",
)
folders = await Folder.filter(parent=parent_folder, owner=current_user, is_deleted=False).order_by("name")
else:
# List root folders (folders with no parent)
folders = await Folder.filter(parent=None, owner=current_user, is_deleted=False).order_by("name")
return [await FolderOut.from_tortoise_orm(folder) for folder in folders]
@router.put("/{folder_id}", response_model=FolderOut)
async def update_folder(folder_id: int, folder_in: FolderUpdate, current_user: User = Depends(get_current_user)):
folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
if folder_in.name:
existing_folder = await Folder.get_or_none(
name=folder_in.name, parent_id=folder.parent_id, owner=current_user
)
if existing_folder and existing_folder.id != folder_id:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Folder with this name already exists in the current parent folder",
)
folder.name = folder_in.name
if folder_in.parent_id is not None:
if folder_in.parent_id == folder_id:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot set folder as its own parent")
new_parent_folder = None
if folder_in.parent_id != 0: # 0 could represent moving to root
new_parent_folder = await Folder.get_or_none(id=folder_in.parent_id, owner=current_user, is_deleted=False)
if not new_parent_folder:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="New parent folder not found or does not belong to the current user",
)
folder.parent = new_parent_folder
await folder.save()
await log_activity(current_user, "folder_updated", "folder", folder.id)
return await FolderOut.from_tortoise_orm(folder)
@router.delete("/{folder_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_folder(folder_id: int, current_user: User = Depends(get_current_user)):
folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
folder.is_deleted = True
await folder.save()
await log_activity(current_user, "folder_deleted", "folder", folder.id)
return
@router.post("/{folder_id}/star", response_model=FolderOut)
async def star_folder(folder_id: int, current_user: User = Depends(get_current_user)):
db_folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not db_folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
db_folder.is_starred = True
await db_folder.save()
await log_activity(current_user, "folder_starred", "folder", folder_id)
return await FolderOut.from_tortoise_orm(db_folder)
@router.post("/{folder_id}/unstar", response_model=FolderOut)
async def unstar_folder(folder_id: int, current_user: User = Depends(get_current_user)):
db_folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not db_folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
db_folder.is_starred = False
await db_folder.save()
await log_activity(current_user, "folder_unstarred", "folder", folder_id)
return await FolderOut.from_tortoise_orm(db_folder)
@router.post("/batch", response_model=List[FolderOut])
async def batch_folder_operations(
batch_operation: BatchFolderOperation,
payload: Optional[BatchMoveCopyPayload] = None,
current_user: User = Depends(get_current_user)
):
updated_folders = []
for folder_id in batch_operation.folder_ids:
db_folder = await Folder.get_or_none(id=folder_id, owner=current_user, is_deleted=False)
if not db_folder:
continue # Skip if folder not found or not owned by user
if batch_operation.operation == "delete":
db_folder.is_deleted = True
await db_folder.save()
await log_activity(current_user, "folder_deleted", "folder", folder_id)
updated_folders.append(db_folder)
elif batch_operation.operation == "star":
db_folder.is_starred = True
await db_folder.save()
await log_activity(current_user, "folder_starred", "folder", folder_id)
updated_folders.append(db_folder)
elif batch_operation.operation == "unstar":
db_folder.is_starred = False
await db_folder.save()
await log_activity(current_user, "folder_unstarred", "folder", folder_id)
updated_folders.append(db_folder)
elif batch_operation.operation == "move" and payload and payload.target_folder_id is not None:
target_folder = await Folder.get_or_none(id=payload.target_folder_id, owner=current_user, is_deleted=False)
if not target_folder:
continue
existing_folder = await Folder.get_or_none(
name=db_folder.name, parent=target_folder, owner=current_user, is_deleted=False
)
if existing_folder and existing_folder.id != folder_id:
continue
db_folder.parent = target_folder
await db_folder.save()
await log_activity(current_user, "folder_moved", "folder", folder_id)
updated_folders.append(db_folder)
return [await FolderOut.from_tortoise_orm(f) for f in updated_folders]
+55
View File
@@ -0,0 +1,55 @@
from fastapi import APIRouter, Depends, Query
from typing import List, Optional
from datetime import datetime
from ..auth import get_current_user
from ..models import User, File, Folder
from ..schemas import FileOut, FolderOut
router = APIRouter(
prefix="/search",
tags=["search"],
)
@router.get("/files", response_model=List[FileOut])
async def search_files(
q: str = Query(..., min_length=1, description="Search query"),
file_type: Optional[str] = Query(None, description="Filter by MIME type prefix (e.g., 'image', 'video')"),
min_size: Optional[int] = Query(None, description="Minimum file size in bytes"),
max_size: Optional[int] = Query(None, description="Maximum file size in bytes"),
date_from: Optional[datetime] = Query(None, description="Filter files created after this date"),
date_to: Optional[datetime] = Query(None, description="Filter files created before this date"),
current_user: User = Depends(get_current_user)
):
query = File.filter(owner=current_user, is_deleted=False, name__icontains=q)
if file_type:
query = query.filter(mime_type__istartswith=file_type)
if min_size is not None:
query = query.filter(size__gte=min_size)
if max_size is not None:
query = query.filter(size__lte=max_size)
if date_from:
query = query.filter(created_at__gte=date_from)
if date_to:
query = query.filter(created_at__lte=date_to)
files = await query.order_by("-created_at").limit(100)
return [await FileOut.from_tortoise_orm(f) for f in files]
@router.get("/folders", response_model=List[FolderOut])
async def search_folders(
q: str = Query(..., min_length=1, description="Search query"),
current_user: User = Depends(get_current_user)
):
folders = await Folder.filter(
owner=current_user,
is_deleted=False,
name__icontains=q
).order_by("-created_at").limit(100)
return [await FolderOut.from_tortoise_orm(folder) for folder in folders]
+226
View File
@@ -0,0 +1,226 @@
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.responses import StreamingResponse
from typing import Optional, List
import secrets
from datetime import datetime, timedelta
from ..auth import get_current_user
from ..models import User, File, Folder, Share
from ..schemas import ShareCreate, ShareOut, FileOut, FolderOut
from ..auth import get_password_hash, verify_password
from ..storage import storage_manager
from ..mail import send_email
from ..settings import settings
router = APIRouter(
prefix="/shares",
tags=["shares"],
)
@router.post("/", response_model=ShareOut, status_code=status.HTTP_201_CREATED)
async def create_share_link(share_in: ShareCreate, current_user: User = Depends(get_current_user)):
if not share_in.file_id and not share_in.folder_id:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Either file_id or folder_id must be provided")
if share_in.file_id and share_in.folder_id:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot share both a file and a folder simultaneously")
file = None
folder = None
if share_in.file_id:
file = await File.get_or_none(id=share_in.file_id, owner=current_user, is_deleted=False)
if not file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found or does not belong to you")
if share_in.folder_id:
folder = await Folder.get_or_none(id=share_in.folder_id, owner=current_user, is_deleted=False)
if not folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found or does not belong to you")
token = secrets.token_urlsafe(16)
hashed_password = None
password_protected = False
if share_in.password:
hashed_password = get_password_hash(share_in.password)
password_protected = True
share = await Share.create(
token=token,
file=file,
folder=folder,
owner=current_user,
expires_at=share_in.expires_at,
password_protected=password_protected,
hashed_password=hashed_password,
permission_level=share_in.permission_level,
)
if share_in.invite_email:
share_url = f"https://{settings.DOMAIN_NAME}/share/{token}"
item_type = "file" if file else "folder"
item_name = file.name if file else folder.name
expiry_text = f" until {share_in.expires_at.strftime('%Y-%m-%d %H:%M')}" if share_in.expires_at else ""
password_text = f"\n\nPassword: {share_in.password}" if share_in.password else ""
email_body = f"""Hello,
{current_user.username} has shared a {item_type} with you: {item_name}
Permission level: {share_in.permission_level}
Access link: {share_url}{password_text}
This link is valid{expiry_text}.
--
MyWebdav File Sharing Service"""
email_html = f"""
<html>
<body>
<p>Hello,</p>
<p><strong>{current_user.username}</strong> has shared a {item_type} with you: <strong>{item_name}</strong></p>
<p><strong>Permission level:</strong> {share_in.permission_level}</p>
<p><a href="{share_url}" style="background-color: #4CAF50; color: white; padding: 10px 20px; text-decoration: none; border-radius: 4px; display: inline-block;">Access {item_type.capitalize()}</a></p>
{f'<p><strong>Password:</strong> {share_in.password}</p>' if share_in.password else ''}
<p><small>This link is valid{expiry_text}.</small></p>
<hr>
<p><small>MyWebdav File Sharing Service</small></p>
</body>
</html>
"""
try:
await send_email(
to_email=share_in.invite_email,
subject=f"{current_user.username} shared {item_name} with you",
body=email_body,
html=email_html
)
except Exception as e:
print(f"Failed to send invitation email: {e}")
return await ShareOut.from_tortoise_orm(share)
@router.get("/my", response_model=List[ShareOut])
async def list_my_shares(current_user: User = Depends(get_current_user)):
shares = await Share.filter(owner=current_user).order_by("-created_at")
return [await ShareOut.from_tortoise_orm(share) for share in shares]
@router.get("/{share_token}", response_model=ShareOut)
async def get_share_link_info(share_token: str):
share = await Share.get_or_none(token=share_token)
if not share:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share link not found")
if share.expires_at and share.expires_at < datetime.utcnow():
raise HTTPException(status_code=status.HTTP_410_GONE, detail="Share link has expired")
# Increment access count
share.access_count += 1
await share.save()
return await ShareOut.from_tortoise_orm(share)
@router.put("/{share_id}", response_model=ShareOut)
async def update_share(share_id: int, share_in: ShareCreate, current_user: User = Depends(get_current_user)):
share = await Share.get_or_none(id=share_id, owner=current_user)
if not share:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share link not found or does not belong to you")
if share_in.expires_at is not None:
share.expires_at = share_in.expires_at
if share_in.password is not None:
share.hashed_password = get_password_hash(share_in.password)
share.password_protected = True
elif share_in.password == "": # Allow clearing password
share.hashed_password = None
share.password_protected = False
if share_in.permission_level is not None:
share.permission_level = share_in.permission_level
await share.save()
return await ShareOut.from_tortoise_orm(share)
@router.post("/{share_token}/access")
async def access_shared_content(share_token: str, password: Optional[str] = None):
share = await Share.get_or_none(token=share_token)
if not share:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share link not found")
if share.expires_at and share.expires_at < datetime.utcnow():
raise HTTPException(status_code=status.HTTP_410_GONE, detail="Share link has expired")
if share.password_protected:
if not password or not verify_password(password, share.hashed_password):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect password")
result = {"message": "Access granted", "permission_level": share.permission_level}
if share.file_id:
file = await File.get_or_none(id=share.file_id, is_deleted=False)
if not file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
result["file"] = await FileOut.from_tortoise_orm(file)
result["type"] = "file"
elif share.folder_id:
folder = await Folder.get_or_none(id=share.folder_id, is_deleted=False)
if not folder:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Folder not found")
result["folder"] = await FolderOut.from_tortoise_orm(folder)
result["type"] = "folder"
files = await File.filter(parent=folder, is_deleted=False)
subfolders = await Folder.filter(parent=folder, is_deleted=False)
result["files"] = [await FileOut.from_tortoise_orm(f) for f in files]
result["folders"] = [await FolderOut.from_tortoise_orm(f) for f in subfolders]
return result
@router.get("/{share_token}/download")
async def download_shared_file(share_token: str, password: Optional[str] = None):
share = await Share.get_or_none(token=share_token)
if not share:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share link not found")
if share.expires_at and share.expires_at < datetime.utcnow():
raise HTTPException(status_code=status.HTTP_410_GONE, detail="Share link has expired")
if share.password_protected:
if not password or not verify_password(password, share.hashed_password):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect password")
if not share.file_id:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="This share is not for a file")
file = await File.get_or_none(id=share.file_id, is_deleted=False)
if not file:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
owner = await User.get(id=file.owner_id)
try:
async def file_iterator():
async for chunk in storage_manager.get_file(owner.id, file.path):
yield chunk
return StreamingResponse(
content=file_iterator(),
media_type=file.mime_type,
headers={
"Content-Disposition": f'attachment; filename="{file.name}"'
}
)
except FileNotFoundError:
raise HTTPException(status_code=404, detail="File not found in storage")
@router.delete("/{share_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_share_link(share_id: int, current_user: User = Depends(get_current_user)):
share = await Share.get_or_none(id=share_id, owner=current_user)
if not share:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share link not found or does not belong to you")
await share.delete()
return
+28
View File
@@ -0,0 +1,28 @@
from typing import List
from fastapi import APIRouter, Depends
from ..auth import get_current_user
from ..models import User, File, Folder
from ..schemas import FileOut, FolderOut
router = APIRouter(
prefix="/starred",
tags=["starred"],
)
@router.get("/files", response_model=List[FileOut])
async def list_starred_files(current_user: User = Depends(get_current_user)):
files = await File.filter(owner=current_user, is_starred=True, is_deleted=False).order_by("name")
return [await FileOut.from_tortoise_orm(f) for f in files]
@router.get("/folders", response_model=List[FolderOut])
async def list_starred_folders(current_user: User = Depends(get_current_user)):
folders = await Folder.filter(owner=current_user, is_starred=True, is_deleted=False).order_by("name")
return [await FolderOut.from_tortoise_orm(f) for f in folders]
@router.get("/all", response_model=List[FileOut]) # This will return files and folders as files for now
async def list_all_starred(current_user: User = Depends(get_current_user)):
starred_files = await File.filter(owner=current_user, is_starred=True, is_deleted=False).order_by("name")
# For simplicity, we'll return files only for now. A more complex solution would involve a union or a custom schema.
return [await FileOut.from_tortoise_orm(f) for f in starred_files]
+52
View File
@@ -0,0 +1,52 @@
from fastapi import APIRouter, Depends
from ..auth import get_current_user
from ..models import User_Pydantic, User, File, Folder
from typing import List, Dict, Any
router = APIRouter(
prefix="/users",
tags=["users"],
)
@router.get("/me", response_model=User_Pydantic)
async def read_users_me(current_user: User = Depends(get_current_user)):
return await User_Pydantic.from_tortoise_orm(current_user)
@router.get("/me/export", response_model=Dict[str, Any])
async def export_my_data(current_user: User = Depends(get_current_user)):
"""
Exports all personal data associated with the current user.
Includes user profile, and metadata for all owned files and folders.
"""
user_data = await User_Pydantic.from_tortoise_orm(current_user)
files = await File.filter(owner=current_user).values(
"id", "name", "size", "created_at", "modified_at", "file_type", "parent_id"
)
folders = await Folder.filter(owner=current_user).values(
"id", "name", "created_at", "modified_at", "parent_id"
)
return {
"user_profile": user_data.dict(),
"files_metadata": files,
"folders_metadata": folders,
# In a more complete implementation, other data like activity logs,
# share information, etc., would also be included.
}
@router.delete("/me", status_code=204)
async def delete_my_account(current_user: User = Depends(get_current_user)):
"""
Deletes the current user's account and all associated data.
This includes all files and folders owned by the user.
"""
# Delete all files and folders owned by the user
await File.filter(owner=current_user).delete()
await Folder.filter(owner=current_user).delete()
# Finally, delete the user account
await current_user.delete()
return {}
+117
View File
@@ -0,0 +1,117 @@
from datetime import datetime
from pydantic import BaseModel, EmailStr
from typing import Optional, List
from tortoise.contrib.pydantic import pydantic_model_creator
class UserCreate(BaseModel):
username: str
email: EmailStr
password: str
class UserLogin(BaseModel):
username: str
password: str
class UserLoginWith2FA(UserLogin):
two_factor_code: Optional[str] = None
class UserAdminUpdate(BaseModel):
username: Optional[str] = None
email: Optional[EmailStr] = None
password: Optional[str] = None
is_active: Optional[bool] = None
is_superuser: Optional[bool] = None
storage_quota_bytes: Optional[int] = None
plan_type: Optional[str] = None
is_2fa_enabled: Optional[bool] = None
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: str | None = None
two_factor_verified: bool = False
class FolderCreate(BaseModel):
name: str
parent_id: Optional[int] = None
class FolderUpdate(BaseModel):
name: Optional[str] = None
parent_id: Optional[int] = None
class ShareCreate(BaseModel):
file_id: Optional[int] = None
folder_id: Optional[int] = None
expires_at: Optional[datetime] = None
password: Optional[str] = None
permission_level: str = "viewer"
invite_email: Optional[EmailStr] = None
class TeamCreate(BaseModel):
name: str
class TeamOut(BaseModel):
id: int
name: str
owner_id: int
created_at: datetime
class Config:
from_attributes = True
class ActivityOut(BaseModel):
id: int
user_id: Optional[int] = None
action: str
target_type: str
target_id: int
ip_address: Optional[str] = None
timestamp: datetime
class Config:
from_attributes = True
class FileRequestCreate(BaseModel):
title: str
description: Optional[str] = None
target_folder_id: int
expires_at: Optional[datetime] = None
class FileRequestOut(BaseModel):
id: int
title: str
description: Optional[str] = None
token: str
owner_id: int
target_folder_id: int
created_at: datetime
expires_at: Optional[datetime] = None
is_active: bool
class Config:
from_attributes = True
from mywebdav.models import Folder, File, Share, FileVersion
FolderOut = pydantic_model_creator(Folder, name="FolderOut")
FileOut = pydantic_model_creator(File, name="FileOut")
ShareOut = pydantic_model_creator(Share, name="ShareOut")
FileVersionOut = pydantic_model_creator(FileVersion, name="FileVersionOut")
class ErrorResponse(BaseModel):
code: int
message: str
details: Optional[str] = None
class BatchFileOperation(BaseModel):
file_ids: List[int]
operation: str # e.g., "delete", "move", "copy", "star", "unstar"
class BatchFolderOperation(BaseModel):
folder_ids: List[int]
operation: str # e.g., "delete", "move", "star", "unstar"
class BatchMoveCopyPayload(BaseModel):
target_folder_id: Optional[int] = None
+37
View File
@@ -0,0 +1,37 @@
import os
import sys
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file='.env', extra='ignore')
DATABASE_URL: str = "sqlite:///app/mywebdav.db"
REDIS_URL: str = "redis://redis:6379/0"
SECRET_KEY: str = "super_secret_key"
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
REFRESH_TOKEN_EXPIRE_DAYS: int = 7
DOMAIN_NAME: str = "MyWebdav.eu"
CERTBOT_EMAIL: str = "admin@example.com"
STORAGE_PATH: str = "storage"
S3_ACCESS_KEY_ID: str | None = None
S3_SECRET_ACCESS_KEY: str | None = None
S3_ENDPOINT_URL: str | None = None
S3_BUCKET_NAME: str = "mywebdav-storage"
SMTP_HOST: str = "mail.example.com"
SMTP_PORT: int = 587
SMTP_USERNAME: str = "noreply@example.com"
SMTP_PASSWORD: str = "your-smtp-password"
SMTP_USE_TLS: bool = True
SMTP_SENDER_EMAIL: str = "noreply@example.com"
TOTP_ISSUER: str = "MyWebdav"
STRIPE_SECRET_KEY: str = ""
STRIPE_PUBLISHABLE_KEY: str = ""
STRIPE_WEBHOOK_SECRET: str = ""
BILLING_ENABLED: bool = False
settings = Settings()
if settings.SECRET_KEY == "super_secret_key" and os.getenv("ENVIRONMENT") == "production":
print("ERROR: Secret key must be changed in production. Set SECRET_KEY environment variable.")
sys.exit(1)
+55
View File
@@ -0,0 +1,55 @@
import os
import aiofiles
from pathlib import Path
from typing import AsyncGenerator
from .settings import settings
class StorageManager:
def __init__(self, base_path: str = settings.STORAGE_PATH):
self.base_path = Path(base_path)
self.base_path.mkdir(parents=True, exist_ok=True)
async def _get_full_path(self, user_id: int, file_path: str) -> Path:
# Ensure file_path is relative and safe
relative_path = Path(file_path).relative_to('/') if str(file_path).startswith('/') else Path(file_path)
full_path = self.base_path / str(user_id) / relative_path
full_path.parent.mkdir(parents=True, exist_ok=True)
return full_path
async def save_file(self, user_id: int, file_path: str, file_content: bytes):
full_path = await self._get_full_path(user_id, file_path)
async with aiofiles.open(full_path, "wb") as f:
await f.write(file_content)
return str(full_path)
async def get_file(self, user_id: int, file_path: str) -> AsyncGenerator:
full_path = await self._get_full_path(user_id, file_path)
if not full_path.exists():
raise FileNotFoundError(f"File not found: {file_path}")
async with aiofiles.open(full_path, "rb") as f:
while chunk := await f.read(8192):
yield chunk
async def delete_file(self, user_id: int, file_path: str):
full_path = await self._get_full_path(user_id, file_path)
if full_path.exists():
os.remove(full_path)
parent_dir = full_path.parent
while parent_dir != self.base_path and parent_dir.exists():
try:
if not any(parent_dir.iterdir()):
parent_dir.rmdir()
parent_dir = parent_dir.parent
else:
break
except OSError:
break
async def file_exists(self, user_id: int, file_path: str) -> bool:
full_path = await self._get_full_path(user_id, file_path)
return full_path.exists()
storage_manager = StorageManager()
+94
View File
@@ -0,0 +1,94 @@
import os
import asyncio
from pathlib import Path
from PIL import Image
import subprocess
from typing import Optional
from .settings import settings
THUMBNAIL_SIZE = (300, 300)
THUMBNAIL_DIR = "thumbnails"
async def generate_thumbnail(file_path: str, mime_type: str, user_id: int) -> Optional[str]:
try:
if mime_type.startswith("image/"):
return await generate_image_thumbnail(file_path, user_id)
elif mime_type.startswith("video/"):
return await generate_video_thumbnail(file_path, user_id)
return None
except Exception as e:
print(f"Error generating thumbnail for {file_path}: {e}")
return None
async def generate_image_thumbnail(file_path: str, user_id: int) -> Optional[str]:
loop = asyncio.get_event_loop()
def _generate():
base_path = Path(settings.STORAGE_PATH)
thumbnail_dir = base_path / str(user_id) / THUMBNAIL_DIR
thumbnail_dir.mkdir(parents=True, exist_ok=True)
file_name = Path(file_path).name
thumbnail_name = f"thumb_{file_name}"
if not thumbnail_name.lower().endswith(('.jpg', '.jpeg', '.png')):
thumbnail_name += ".jpg"
thumbnail_path = thumbnail_dir / thumbnail_name
actual_file_path = base_path / str(user_id) / file_path if not Path(file_path).is_absolute() else Path(file_path)
with Image.open(actual_file_path) as img:
img.thumbnail(THUMBNAIL_SIZE, Image.Resampling.LANCZOS)
if img.mode in ("RGBA", "LA", "P"):
background = Image.new("RGB", img.size, (255, 255, 255))
if img.mode == "P":
img = img.convert("RGBA")
background.paste(img, mask=img.split()[-1] if img.mode in ("RGBA", "LA") else None)
img = background
img.save(str(thumbnail_path), "JPEG", quality=85, optimize=True)
return str(thumbnail_path.relative_to(base_path / str(user_id)))
return await loop.run_in_executor(None, _generate)
async def generate_video_thumbnail(file_path: str, user_id: int) -> Optional[str]:
loop = asyncio.get_event_loop()
def _generate():
base_path = Path(settings.STORAGE_PATH)
thumbnail_dir = base_path / str(user_id) / THUMBNAIL_DIR
thumbnail_dir.mkdir(parents=True, exist_ok=True)
file_name = Path(file_path).stem
thumbnail_name = f"thumb_{file_name}.jpg"
thumbnail_path = thumbnail_dir / thumbnail_name
actual_file_path = base_path / str(user_id) / file_path if not Path(file_path).is_absolute() else Path(file_path)
subprocess.run([
"ffmpeg",
"-i", str(actual_file_path),
"-ss", "00:00:01",
"-vframes", "1",
"-vf", f"scale={THUMBNAIL_SIZE[0]}:{THUMBNAIL_SIZE[1]}:force_original_aspect_ratio=decrease",
"-y",
str(thumbnail_path)
], check=True, capture_output=True)
return str(thumbnail_path.relative_to(base_path / str(user_id)))
try:
return await loop.run_in_executor(None, _generate)
except subprocess.CalledProcessError:
return None
async def delete_thumbnail(thumbnail_path: str, user_id: int):
try:
base_path = Path(settings.STORAGE_PATH)
full_path = base_path / str(user_id) / thumbnail_path
if full_path.exists():
full_path.unlink()
except Exception as e:
print(f"Error deleting thumbnail {thumbnail_path}: {e}")
+60
View File
@@ -0,0 +1,60 @@
import pyotp
import qrcode
import io
import base64
import secrets
import hashlib
from typing import List, Optional
def generate_totp_secret() -> str:
"""Generates a random base32 TOTP secret."""
return pyotp.random_base32()
def generate_totp_uri(secret: str, account_name: str, issuer_name: str) -> str:
"""Generates a Google Authenticator-compatible TOTP URI."""
return pyotp.totp.TOTP(secret).provisioning_uri(name=account_name, issuer_name=issuer_name)
def generate_qr_code_base64(uri: str) -> str:
"""Generates a base64 encoded QR code image for a given URI."""
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=10,
border=4,
)
qr.add_data(uri)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
buffered = io.BytesIO()
img.save(buffered, format="PNG")
return base64.b64encode(buffered.getvalue()).decode("utf-8")
def verify_totp_code(secret: str, code: str) -> bool:
"""Verifies a TOTP code against a secret."""
totp = pyotp.TOTP(secret)
return totp.verify(code)
def generate_recovery_codes(num_codes: int = 10) -> List[str]:
"""Generates a list of random recovery codes."""
return [secrets.token_urlsafe(16) for _ in range(num_codes)]
def hash_recovery_code(code: str) -> str:
"""Hashes a single recovery code using SHA256."""
return hashlib.sha256(code.encode('utf-8')).hexdigest()
def verify_recovery_code(plain_code: str, hashed_code: str) -> bool:
"""Verifies a plain recovery code against its hashed version."""
return hash_recovery_code(plain_code) == hashed_code
def hash_recovery_codes(codes: List[str]) -> List[str]:
"""Hashes a list of recovery codes."""
return [hash_recovery_code(code) for code in codes]
def verify_recovery_codes(plain_code: str, hashed_codes: List[str]) -> bool:
"""Verifies if a plain recovery code matches any of the hashed recovery codes."""
for hashed_code in hashed_codes:
if verify_recovery_code(plain_code, hashed_code):
return True
return False
+840
View File
@@ -0,0 +1,840 @@
from fastapi import APIRouter, Request, Response, Depends, HTTPException, status, Header
from fastapi.responses import StreamingResponse
from typing import Optional
from xml.etree import ElementTree as ET
from datetime import datetime
import hashlib
import mimetypes
import os
import base64
from urllib.parse import unquote, urlparse
from .auth import get_current_user, verify_password
from .models import User, File, Folder, WebDAVProperty
from .storage import storage_manager
from .activity import log_activity
router = APIRouter(
prefix="/webdav",
tags=["webdav"],
)
class WebDAVLock:
locks = {}
@classmethod
def create_lock(cls, path: str, user_id: int, timeout: int = 3600):
lock_token = f"opaquelocktoken:{hashlib.md5(f'{path}{user_id}{datetime.now()}'.encode()).hexdigest()}"
cls.locks[path] = {
'token': lock_token,
'user_id': user_id,
'created_at': datetime.now(),
'timeout': timeout
}
return lock_token
@classmethod
def get_lock(cls, path: str):
return cls.locks.get(path)
@classmethod
def remove_lock(cls, path: str):
if path in cls.locks:
del cls.locks[path]
async def basic_auth(authorization: Optional[str] = Header(None)):
if not authorization:
return None
try:
scheme, credentials = authorization.split()
if scheme.lower() != 'basic':
return None
decoded = base64.b64decode(credentials).decode('utf-8')
username, password = decoded.split(':', 1)
user = await User.get_or_none(username=username)
if user and verify_password(password, user.hashed_password):
return user
except (ValueError, UnicodeDecodeError, base64.binascii.Error):
return None
return None
async def webdav_auth(request: Request, authorization: Optional[str] = Header(None)):
user = await basic_auth(authorization)
if user:
return user
try:
user = await get_current_user(request)
return user
except HTTPException:
pass
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
headers={'WWW-Authenticate': 'Basic realm="MyWebdav WebDAV"'}
)
async def resolve_path(path_str: str, user: User):
if not path_str or path_str == '/':
return None, None, True
parts = [p for p in path_str.split('/') if p]
if not parts:
return None, None, True
current_folder = None
for i, part in enumerate(parts[:-1]):
folder = await Folder.get_or_none(
name=part,
parent=current_folder,
owner=user,
is_deleted=False
)
if not folder:
return None, None, False
current_folder = folder
last_part = parts[-1]
folder = await Folder.get_or_none(
name=last_part,
parent=current_folder,
owner=user,
is_deleted=False
)
if folder:
return folder, current_folder, True
file = await File.get_or_none(
name=last_part,
parent=current_folder,
owner=user,
is_deleted=False
)
if file:
return file, current_folder, True
return None, current_folder, True
def build_href(base_path: str, name: str, is_collection: bool):
path = f"{base_path.rstrip('/')}/{name}"
if is_collection:
path += '/'
return path
async def get_custom_properties(resource_type: str, resource_id: int):
props = await WebDAVProperty.filter(
resource_type=resource_type,
resource_id=resource_id
)
return {(prop.namespace, prop.name): prop.value for prop in props}
def create_propstat_element(props: dict, custom_props: dict = None, status: str = "HTTP/1.1 200 OK"):
propstat = ET.Element("D:propstat")
prop = ET.SubElement(propstat, "D:prop")
for key, value in props.items():
if key == "resourcetype":
resourcetype = ET.SubElement(prop, "D:resourcetype")
if value == "collection":
ET.SubElement(resourcetype, "D:collection")
elif key == "getcontentlength":
elem = ET.SubElement(prop, f"D:{key}")
elem.text = str(value)
elif key == "getcontenttype":
elem = ET.SubElement(prop, f"D:{key}")
elem.text = value
elif key == "getlastmodified":
elem = ET.SubElement(prop, f"D:{key}")
elem.text = value.strftime('%a, %d %b %Y %H:%M:%S GMT')
elif key == "creationdate":
elem = ET.SubElement(prop, f"D:{key}")
elem.text = value.isoformat() + 'Z'
elif key == "displayname":
elem = ET.SubElement(prop, f"D:{key}")
elem.text = value
elif key == "getetag":
elem = ET.SubElement(prop, f"D:{key}")
elem.text = value
if custom_props:
for (namespace, name), value in custom_props.items():
if namespace == "DAV:":
continue
elem = ET.SubElement(prop, f"{{{namespace}}}{name}")
elem.text = value
status_elem = ET.SubElement(propstat, "D:status")
status_elem.text = status
return propstat
def parse_propfind_body(body: bytes):
if not body:
return None
try:
root = ET.fromstring(body)
allprop = root.find(".//{DAV:}allprop")
if allprop is not None:
return "allprop"
propname = root.find(".//{DAV:}propname")
if propname is not None:
return "propname"
prop = root.find(".//{DAV:}prop")
if prop is not None:
requested_props = []
for child in prop:
ns = child.tag.split('}')[0][1:] if '}' in child.tag else "DAV:"
name = child.tag.split('}')[1] if '}' in child.tag else child.tag
requested_props.append((ns, name))
return requested_props
except ET.ParseError:
return None
return None
@router.api_route("/{full_path:path}", methods=["OPTIONS"])
async def webdav_options(full_path: str):
return Response(
status_code=200,
headers={
"DAV": "1, 2",
"Allow": "OPTIONS, GET, HEAD, POST, PUT, DELETE, PROPFIND, PROPPATCH, MKCOL, COPY, MOVE, LOCK, UNLOCK",
"MS-Author-Via": "DAV"
}
)
@router.api_route("/{full_path:path}", methods=["PROPFIND"])
async def handle_propfind(request: Request, full_path: str, current_user: User = Depends(webdav_auth)):
depth = request.headers.get("Depth", "1")
full_path = unquote(full_path).strip('/')
body = await request.body()
requested_props = parse_propfind_body(body)
resource, parent, exists = await resolve_path(full_path, current_user)
if not exists and resource is None:
raise HTTPException(status_code=404, detail="Not found")
multistatus = ET.Element("D:multistatus", {"xmlns:D": "DAV:"})
base_href = f"/webdav/{full_path}" if full_path else "/webdav/"
if resource is None:
response = ET.SubElement(multistatus, "D:response")
href = ET.SubElement(response, "D:href")
href.text = base_href if base_href.endswith('/') else base_href + '/'
props = {
"resourcetype": "collection",
"displayname": full_path.split('/')[-1] if full_path else "Root",
"creationdate": datetime.now(),
"getlastmodified": datetime.now()
}
response.append(create_propstat_element(props))
if depth in ["1", "infinity"]:
folders = await Folder.filter(owner=current_user, parent=parent, is_deleted=False)
files = await File.filter(owner=current_user, parent=parent, is_deleted=False)
for folder in folders:
response = ET.SubElement(multistatus, "D:response")
href = ET.SubElement(response, "D:href")
href.text = build_href(base_href, folder.name, True)
props = {
"resourcetype": "collection",
"displayname": folder.name,
"creationdate": folder.created_at,
"getlastmodified": folder.updated_at
}
custom_props = await get_custom_properties("folder", folder.id) if requested_props == "allprop" or (isinstance(requested_props, list) and any(ns != "DAV:" for ns, _ in requested_props)) else None
response.append(create_propstat_element(props, custom_props))
for file in files:
response = ET.SubElement(multistatus, "D:response")
href = ET.SubElement(response, "D:href")
href.text = build_href(base_href, file.name, False)
props = {
"resourcetype": "",
"displayname": file.name,
"getcontentlength": file.size,
"getcontenttype": file.mime_type,
"creationdate": file.created_at,
"getlastmodified": file.updated_at,
"getetag": f'"{file.file_hash}"'
}
custom_props = await get_custom_properties("file", file.id) if requested_props == "allprop" or (isinstance(requested_props, list) and any(ns != "DAV:" for ns, _ in requested_props)) else None
response.append(create_propstat_element(props, custom_props))
elif isinstance(resource, Folder):
response = ET.SubElement(multistatus, "D:response")
href = ET.SubElement(response, "D:href")
href.text = base_href if base_href.endswith('/') else base_href + '/'
props = {
"resourcetype": "collection",
"displayname": resource.name,
"creationdate": resource.created_at,
"getlastmodified": resource.updated_at
}
custom_props = await get_custom_properties("folder", resource.id) if requested_props == "allprop" or (isinstance(requested_props, list) and any(ns != "DAV:" for ns, _ in requested_props)) else None
response.append(create_propstat_element(props, custom_props))
if depth in ["1", "infinity"]:
folders = await Folder.filter(owner=current_user, parent=resource, is_deleted=False)
files = await File.filter(owner=current_user, parent=resource, is_deleted=False)
for folder in folders:
response = ET.SubElement(multistatus, "D:response")
href = ET.SubElement(response, "D:href")
href.text = build_href(base_href, folder.name, True)
props = {
"resourcetype": "collection",
"displayname": folder.name,
"creationdate": folder.created_at,
"getlastmodified": folder.updated_at
}
custom_props = await get_custom_properties("folder", folder.id) if requested_props == "allprop" or (isinstance(requested_props, list) and any(ns != "DAV:" for ns, _ in requested_props)) else None
response.append(create_propstat_element(props, custom_props))
for file in files:
response = ET.SubElement(multistatus, "D:response")
href = ET.SubElement(response, "D:href")
href.text = build_href(base_href, file.name, False)
props = {
"resourcetype": "",
"displayname": file.name,
"getcontentlength": file.size,
"getcontenttype": file.mime_type,
"creationdate": file.created_at,
"getlastmodified": file.updated_at,
"getetag": f'"{file.file_hash}"'
}
custom_props = await get_custom_properties("file", file.id) if requested_props == "allprop" or (isinstance(requested_props, list) and any(ns != "DAV:" for ns, _ in requested_props)) else None
response.append(create_propstat_element(props, custom_props))
elif isinstance(resource, File):
response = ET.SubElement(multistatus, "D:response")
href = ET.SubElement(response, "D:href")
href.text = base_href
props = {
"resourcetype": "",
"displayname": resource.name,
"getcontentlength": resource.size,
"getcontenttype": resource.mime_type,
"creationdate": resource.created_at,
"getlastmodified": resource.updated_at,
"getetag": f'"{resource.file_hash}"'
}
custom_props = await get_custom_properties("file", resource.id) if requested_props == "allprop" or (isinstance(requested_props, list) and any(ns != "DAV:" for ns, _ in requested_props)) else None
response.append(create_propstat_element(props, custom_props))
xml_content = ET.tostring(multistatus, encoding="utf-8", xml_declaration=True)
return Response(content=xml_content, media_type="application/xml; charset=utf-8", status_code=207)
@router.api_route("/{full_path:path}", methods=["GET", "HEAD"])
async def handle_get(request: Request, full_path: str, current_user: User = Depends(webdav_auth)):
full_path = unquote(full_path).strip('/')
resource, parent, exists = await resolve_path(full_path, current_user)
if not isinstance(resource, File):
raise HTTPException(status_code=404, detail="File not found")
try:
if request.method == "HEAD":
return Response(
status_code=200,
headers={
"Content-Length": str(resource.size),
"Content-Type": resource.mime_type,
"ETag": f'"{resource.file_hash}"',
"Last-Modified": resource.updated_at.strftime('%a, %d %b %Y %H:%M:%S GMT')
}
)
async def file_iterator():
async for chunk in storage_manager.get_file(current_user.id, resource.path):
yield chunk
return StreamingResponse(
content=file_iterator(),
media_type=resource.mime_type,
headers={
"Content-Disposition": f'attachment; filename="{resource.name}"',
"ETag": f'"{resource.file_hash}"',
"Last-Modified": resource.updated_at.strftime('%a, %d %b %Y %H:%M:%S GMT')
}
)
except FileNotFoundError:
raise HTTPException(status_code=404, detail="File not found in storage")
@router.api_route("/{full_path:path}", methods=["PUT"])
async def handle_put(request: Request, full_path: str, current_user: User = Depends(webdav_auth)):
full_path = unquote(full_path).strip('/')
if not full_path:
raise HTTPException(status_code=400, detail="Cannot PUT to root")
parts = [p for p in full_path.split('/') if p]
file_name = parts[-1]
parent_path = '/'.join(parts[:-1]) if len(parts) > 1 else ''
_, parent_folder, exists = await resolve_path(parent_path, current_user)
if not exists:
raise HTTPException(status_code=409, detail="Parent folder does not exist")
file_content = await request.body()
file_size = len(file_content)
if current_user.used_storage_bytes + file_size > current_user.storage_quota_bytes:
raise HTTPException(status_code=507, detail="Storage quota exceeded")
file_hash = hashlib.sha256(file_content).hexdigest()
file_extension = os.path.splitext(file_name)[1]
unique_filename = f"{file_hash}{file_extension}"
storage_path = os.path.join(str(current_user.id), unique_filename)
await storage_manager.save_file(current_user.id, storage_path, file_content)
mime_type, _ = mimetypes.guess_type(file_name)
if not mime_type:
mime_type = "application/octet-stream"
existing_file = await File.get_or_none(
name=file_name,
parent=parent_folder,
owner=current_user,
is_deleted=False
)
if existing_file:
old_size = existing_file.size
existing_file.path = storage_path
existing_file.size = file_size
existing_file.mime_type = mime_type
existing_file.file_hash = file_hash
existing_file.updated_at = datetime.now()
await existing_file.save()
current_user.used_storage_bytes = current_user.used_storage_bytes - old_size + file_size
await current_user.save()
await log_activity(current_user, "file_updated", "file", existing_file.id)
return Response(status_code=204)
else:
db_file = await File.create(
name=file_name,
path=storage_path,
size=file_size,
mime_type=mime_type,
file_hash=file_hash,
owner=current_user,
parent=parent_folder
)
current_user.used_storage_bytes += file_size
await current_user.save()
await log_activity(current_user, "file_created", "file", db_file.id)
return Response(status_code=201)
@router.api_route("/{full_path:path}", methods=["DELETE"])
async def handle_delete(request: Request, full_path: str, current_user: User = Depends(webdav_auth)):
full_path = unquote(full_path).strip('/')
if not full_path:
raise HTTPException(status_code=400, detail="Cannot DELETE root")
resource, parent, exists = await resolve_path(full_path, current_user)
if not resource:
raise HTTPException(status_code=404, detail="Resource not found")
if isinstance(resource, File):
resource.is_deleted = True
resource.deleted_at = datetime.now()
await resource.save()
await log_activity(current_user, "file_deleted", "file", resource.id)
elif isinstance(resource, Folder):
resource.is_deleted = True
await resource.save()
await log_activity(current_user, "folder_deleted", "folder", resource.id)
return Response(status_code=204)
@router.api_route("/{full_path:path}", methods=["MKCOL"])
async def handle_mkcol(request: Request, full_path: str, current_user: User = Depends(webdav_auth)):
full_path = unquote(full_path).strip('/')
if not full_path:
raise HTTPException(status_code=400, detail="Cannot MKCOL at root")
parts = [p for p in full_path.split('/') if p]
folder_name = parts[-1]
parent_path = '/'.join(parts[:-1]) if len(parts) > 1 else ''
_, parent_folder, exists = await resolve_path(parent_path, current_user)
if not exists:
raise HTTPException(status_code=409, detail="Parent folder does not exist")
existing = await Folder.get_or_none(
name=folder_name,
parent=parent_folder,
owner=current_user,
is_deleted=False
)
if existing:
raise HTTPException(status_code=405, detail="Folder already exists")
folder = await Folder.create(
name=folder_name,
parent=parent_folder,
owner=current_user
)
await log_activity(current_user, "folder_created", "folder", folder.id)
return Response(status_code=201)
@router.api_route("/{full_path:path}", methods=["COPY"])
async def handle_copy(request: Request, full_path: str, current_user: User = Depends(webdav_auth)):
full_path = unquote(full_path).strip('/')
destination = request.headers.get("Destination")
overwrite = request.headers.get("Overwrite", "T")
if not destination:
raise HTTPException(status_code=400, detail="Destination header required")
dest_path = unquote(urlparse(destination).path)
dest_path = dest_path.replace('/webdav/', '').strip('/')
source_resource, _, exists = await resolve_path(full_path, current_user)
if not source_resource:
raise HTTPException(status_code=404, detail="Source not found")
if not isinstance(source_resource, File):
raise HTTPException(status_code=501, detail="Only file copy is implemented")
dest_parts = [p for p in dest_path.split('/') if p]
dest_name = dest_parts[-1]
dest_parent_path = '/'.join(dest_parts[:-1]) if len(dest_parts) > 1 else ''
_, dest_parent, exists = await resolve_path(dest_parent_path, current_user)
if not exists:
raise HTTPException(status_code=409, detail="Destination parent does not exist")
existing_dest = await File.get_or_none(
name=dest_name,
parent=dest_parent,
owner=current_user,
is_deleted=False
)
if existing_dest and overwrite == "F":
raise HTTPException(status_code=412, detail="Destination exists and overwrite is false")
if existing_dest:
await existing_dest.delete()
new_file = await File.create(
name=dest_name,
path=source_resource.path,
size=source_resource.size,
mime_type=source_resource.mime_type,
file_hash=source_resource.file_hash,
owner=current_user,
parent=dest_parent
)
await log_activity(current_user, "file_copied", "file", new_file.id)
return Response(status_code=201 if not existing_dest else 204)
@router.api_route("/{full_path:path}", methods=["MOVE"])
async def handle_move(request: Request, full_path: str, current_user: User = Depends(webdav_auth)):
full_path = unquote(full_path).strip('/')
destination = request.headers.get("Destination")
overwrite = request.headers.get("Overwrite", "T")
if not destination:
raise HTTPException(status_code=400, detail="Destination header required")
dest_path = unquote(urlparse(destination).path)
dest_path = dest_path.replace('/webdav/', '').strip('/')
source_resource, _, exists = await resolve_path(full_path, current_user)
if not source_resource:
raise HTTPException(status_code=404, detail="Source not found")
dest_parts = [p for p in dest_path.split('/') if p]
dest_name = dest_parts[-1]
dest_parent_path = '/'.join(dest_parts[:-1]) if len(dest_parts) > 1 else ''
_, dest_parent, exists = await resolve_path(dest_parent_path, current_user)
if not exists:
raise HTTPException(status_code=409, detail="Destination parent does not exist")
if isinstance(source_resource, File):
existing_dest = await File.get_or_none(
name=dest_name,
parent=dest_parent,
owner=current_user,
is_deleted=False
)
if existing_dest and overwrite == "F":
raise HTTPException(status_code=412, detail="Destination exists and overwrite is false")
if existing_dest:
await existing_dest.delete()
source_resource.name = dest_name
source_resource.parent = dest_parent
await source_resource.save()
await log_activity(current_user, "file_moved", "file", source_resource.id)
return Response(status_code=201 if not existing_dest else 204)
elif isinstance(source_resource, Folder):
existing_dest = await Folder.get_or_none(
name=dest_name,
parent=dest_parent,
owner=current_user,
is_deleted=False
)
if existing_dest and overwrite == "F":
raise HTTPException(status_code=412, detail="Destination exists and overwrite is false")
if existing_dest:
existing_dest.is_deleted = True
await existing_dest.save()
source_resource.name = dest_name
source_resource.parent = dest_parent
await source_resource.save()
await log_activity(current_user, "folder_moved", "folder", source_resource.id)
return Response(status_code=201 if not existing_dest else 204)
@router.api_route("/{full_path:path}", methods=["LOCK"])
async def handle_lock(request: Request, full_path: str, current_user: User = Depends(webdav_auth)):
full_path = unquote(full_path).strip('/')
timeout_header = request.headers.get("Timeout", "Second-3600")
timeout = 3600
if timeout_header.startswith("Second-"):
try:
timeout = int(timeout_header.split("-")[1])
except (ValueError, IndexError):
pass
lock_token = WebDAVLock.create_lock(full_path, current_user.id, timeout)
lockinfo = ET.Element("D:prop", {"xmlns:D": "DAV:"})
lockdiscovery = ET.SubElement(lockinfo, "D:lockdiscovery")
activelock = ET.SubElement(lockdiscovery, "D:activelock")
locktype = ET.SubElement(activelock, "D:locktype")
ET.SubElement(locktype, "D:write")
lockscope = ET.SubElement(activelock, "D:lockscope")
ET.SubElement(lockscope, "D:exclusive")
depth_elem = ET.SubElement(activelock, "D:depth")
depth_elem.text = "0"
owner = ET.SubElement(activelock, "D:owner")
owner_href = ET.SubElement(owner, "D:href")
owner_href.text = current_user.username
timeout_elem = ET.SubElement(activelock, "D:timeout")
timeout_elem.text = f"Second-{timeout}"
locktoken_elem = ET.SubElement(activelock, "D:locktoken")
href = ET.SubElement(locktoken_elem, "D:href")
href.text = lock_token
xml_content = ET.tostring(lockinfo, encoding="utf-8", xml_declaration=True)
return Response(
content=xml_content,
media_type="application/xml; charset=utf-8",
status_code=200,
headers={"Lock-Token": f"<{lock_token}>"}
)
@router.api_route("/{full_path:path}", methods=["UNLOCK"])
async def handle_unlock(request: Request, full_path: str, current_user: User = Depends(webdav_auth)):
full_path = unquote(full_path).strip('/')
lock_token_header = request.headers.get("Lock-Token")
if not lock_token_header:
raise HTTPException(status_code=400, detail="Lock-Token header required")
lock_token = lock_token_header.strip('<>')
existing_lock = WebDAVLock.get_lock(full_path)
if not existing_lock or existing_lock['token'] != lock_token:
raise HTTPException(status_code=409, detail="Invalid lock token")
if existing_lock['user_id'] != current_user.id:
raise HTTPException(status_code=403, detail="Not lock owner")
WebDAVLock.remove_lock(full_path)
return Response(status_code=204)
@router.api_route("/{full_path:path}", methods=["PROPPATCH"])
async def handle_proppatch(request: Request, full_path: str, current_user: User = Depends(webdav_auth)):
full_path = unquote(full_path).strip('/')
resource, parent, exists = await resolve_path(full_path, current_user)
if not resource:
raise HTTPException(status_code=404, detail="Resource not found")
body = await request.body()
if not body:
raise HTTPException(status_code=400, detail="Request body required")
try:
root = ET.fromstring(body)
except:
raise HTTPException(status_code=400, detail="Invalid XML")
resource_type = "file" if isinstance(resource, File) else "folder"
resource_id = resource.id
set_props = []
remove_props = []
failed_props = []
set_element = root.find(".//{DAV:}set")
if set_element is not None:
prop_element = set_element.find(".//{DAV:}prop")
if prop_element is not None:
for child in prop_element:
ns = child.tag.split('}')[0][1:] if '}' in child.tag else "DAV:"
name = child.tag.split('}')[1] if '}' in child.tag else child.tag
value = child.text or ""
if ns == "DAV:":
live_props = ["creationdate", "getcontentlength", "getcontenttype",
"getetag", "getlastmodified", "resourcetype"]
if name in live_props:
failed_props.append((ns, name, "409 Conflict"))
continue
try:
existing_prop = await WebDAVProperty.get_or_none(
resource_type=resource_type,
resource_id=resource_id,
namespace=ns,
name=name
)
if existing_prop:
existing_prop.value = value
await existing_prop.save()
else:
await WebDAVProperty.create(
resource_type=resource_type,
resource_id=resource_id,
namespace=ns,
name=name,
value=value
)
set_props.append((ns, name))
except Exception as e:
failed_props.append((ns, name, "500 Internal Server Error"))
remove_element = root.find(".//{DAV:}remove")
if remove_element is not None:
prop_element = remove_element.find(".//{DAV:}prop")
if prop_element is not None:
for child in prop_element:
ns = child.tag.split('}')[0][1:] if '}' in child.tag else "DAV:"
name = child.tag.split('}')[1] if '}' in child.tag else child.tag
if ns == "DAV:":
failed_props.append((ns, name, "409 Conflict"))
continue
try:
existing_prop = await WebDAVProperty.get_or_none(
resource_type=resource_type,
resource_id=resource_id,
namespace=ns,
name=name
)
if existing_prop:
await existing_prop.delete()
remove_props.append((ns, name))
else:
failed_props.append((ns, name, "404 Not Found"))
except Exception as e:
failed_props.append((ns, name, "500 Internal Server Error"))
multistatus = ET.Element("D:multistatus", {"xmlns:D": "DAV:"})
response_elem = ET.SubElement(multistatus, "D:response")
href = ET.SubElement(response_elem, "D:href")
href.text = f"/webdav/{full_path}"
if set_props or remove_props:
propstat = ET.SubElement(response_elem, "D:propstat")
prop = ET.SubElement(propstat, "D:prop")
for ns, name in set_props + remove_props:
if ns == "DAV:":
ET.SubElement(prop, f"D:{name}")
else:
ET.SubElement(prop, f"{{{ns}}}{name}")
status_elem = ET.SubElement(propstat, "D:status")
status_elem.text = "HTTP/1.1 200 OK"
if failed_props:
prop_by_status = {}
for ns, name, status_text in failed_props:
if status_text not in prop_by_status:
prop_by_status[status_text] = []
prop_by_status[status_text].append((ns, name))
for status_text, props_list in prop_by_status.items():
propstat = ET.SubElement(response_elem, "D:propstat")
prop = ET.SubElement(propstat, "D:prop")
for ns, name in props_list:
if ns == "DAV:":
ET.SubElement(prop, f"D:{name}")
else:
ET.SubElement(prop, f"{{{ns}}}{name}")
status_elem = ET.SubElement(propstat, "D:status")
status_elem.text = f"HTTP/1.1 {status_text}"
await log_activity(current_user, "properties_modified", resource_type, resource_id)
xml_content = ET.tostring(multistatus, encoding="utf-8", xml_declaration=True)
return Response(content=xml_content, media_type="application/xml; charset=utf-8", status_code=207)