318 lines
11 KiB
Python
318 lines
11 KiB
Python
import uuid
|
|
import datetime
|
|
import hashlib
|
|
import logging
|
|
from typing import Dict, List, Optional, Any
|
|
from .storage_service import StorageService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class SharingService:
|
|
|
|
PERMISSION_VIEW = "view"
|
|
PERMISSION_EDIT = "edit"
|
|
PERMISSION_COMMENT = "comment"
|
|
|
|
SCOPE_PUBLIC = "public"
|
|
SCOPE_PRIVATE = "private"
|
|
SCOPE_ACCOUNT_BASED = "account_based"
|
|
|
|
def __init__(self):
|
|
self.storage = StorageService()
|
|
|
|
async def _load_shares(self, user_email: str) -> Dict:
|
|
shares = await self.storage.load(user_email, "shares")
|
|
return shares if shares else {}
|
|
|
|
async def _save_shares(self, user_email: str, shares: Dict):
|
|
await self.storage.save(user_email, "shares", shares)
|
|
|
|
async def _load_share_recipients(self, share_id: str) -> Dict:
|
|
recipients = await self.storage.load("global_shares", f"recipients_{share_id}")
|
|
return recipients if recipients else {}
|
|
|
|
async def _save_share_recipients(self, share_id: str, recipients: Dict):
|
|
await self.storage.save("global_shares", f"recipients_{share_id}", recipients)
|
|
|
|
async def _load_all_shares(self) -> Dict:
|
|
all_shares = await self.storage.load("global_shares", "all_shares_index")
|
|
return all_shares if all_shares else {}
|
|
|
|
async def _save_all_shares(self, all_shares: Dict):
|
|
await self.storage.save("global_shares", "all_shares_index", all_shares)
|
|
|
|
def _hash_password(self, password: str) -> str:
|
|
return hashlib.sha256(password.encode()).hexdigest()
|
|
|
|
async def create_share(
|
|
self,
|
|
owner_email: str,
|
|
item_path: str,
|
|
permission: str = PERMISSION_VIEW,
|
|
scope: str = SCOPE_PUBLIC,
|
|
password: Optional[str] = None,
|
|
expiration_days: Optional[int] = None,
|
|
disable_download: bool = False,
|
|
recipient_emails: Optional[List[str]] = None
|
|
) -> str:
|
|
|
|
share_id = str(uuid.uuid4())
|
|
now = datetime.datetime.now(datetime.timezone.utc)
|
|
|
|
expires_at = None
|
|
if expiration_days:
|
|
expires_at = (now + datetime.timedelta(days=expiration_days)).isoformat()
|
|
|
|
password_hash = None
|
|
if password:
|
|
password_hash = self._hash_password(password)
|
|
|
|
share_data = {
|
|
"share_id": share_id,
|
|
"owner_email": owner_email,
|
|
"item_path": item_path,
|
|
"permission": permission,
|
|
"scope": scope,
|
|
"password_hash": password_hash,
|
|
"created_at": now.isoformat(),
|
|
"expires_at": expires_at,
|
|
"disable_download": disable_download,
|
|
"active": True,
|
|
"access_count": 0,
|
|
"last_accessed": None
|
|
}
|
|
|
|
user_shares = await self._load_shares(owner_email)
|
|
user_shares[share_id] = share_data
|
|
await self._save_shares(owner_email, user_shares)
|
|
|
|
all_shares = await self._load_all_shares()
|
|
all_shares[share_id] = {
|
|
"owner_email": owner_email,
|
|
"item_path": item_path,
|
|
"created_at": now.isoformat()
|
|
}
|
|
await self._save_all_shares(all_shares)
|
|
|
|
if recipient_emails and scope in [self.SCOPE_PRIVATE, self.SCOPE_ACCOUNT_BASED]:
|
|
recipients = {}
|
|
for email in recipient_emails:
|
|
recipients[email] = {
|
|
"email": email,
|
|
"permission": permission,
|
|
"invited_at": now.isoformat(),
|
|
"accessed": False
|
|
}
|
|
await self._save_share_recipients(share_id, recipients)
|
|
|
|
logger.info(f"Created share {share_id} for {item_path} by {owner_email}")
|
|
return share_id
|
|
|
|
async def get_share(self, share_id: str) -> Optional[Dict]:
|
|
all_shares = await self._load_all_shares()
|
|
if share_id not in all_shares:
|
|
return None
|
|
|
|
owner_email = all_shares[share_id]["owner_email"]
|
|
user_shares = await self._load_shares(owner_email)
|
|
|
|
if share_id not in user_shares:
|
|
return None
|
|
|
|
share = user_shares[share_id]
|
|
|
|
if not share.get("active", True):
|
|
logger.warning(f"Share {share_id} is deactivated")
|
|
return None
|
|
|
|
if share.get("expires_at"):
|
|
expiry_time = datetime.datetime.fromisoformat(share["expires_at"])
|
|
if expiry_time <= datetime.datetime.now(datetime.timezone.utc):
|
|
logger.warning(f"Share {share_id} has expired")
|
|
return None
|
|
|
|
return share
|
|
|
|
async def verify_share_access(
|
|
self,
|
|
share_id: str,
|
|
password: Optional[str] = None,
|
|
accessor_email: Optional[str] = None
|
|
) -> bool:
|
|
|
|
share = await self.get_share(share_id)
|
|
if not share:
|
|
return False
|
|
|
|
if share.get("password_hash") and password:
|
|
if self._hash_password(password) != share["password_hash"]:
|
|
logger.warning(f"Invalid password for share {share_id}")
|
|
return False
|
|
elif share.get("password_hash") and not password:
|
|
return False
|
|
|
|
if share["scope"] == self.SCOPE_PRIVATE and accessor_email:
|
|
recipients = await self._load_share_recipients(share_id)
|
|
if accessor_email not in recipients:
|
|
logger.warning(f"Email {accessor_email} not in recipients for share {share_id}")
|
|
return False
|
|
|
|
if share["scope"] == self.SCOPE_ACCOUNT_BASED and not accessor_email:
|
|
logger.warning(f"Account-based share {share_id} requires authenticated user")
|
|
return False
|
|
|
|
return True
|
|
|
|
async def record_share_access(self, share_id: str, accessor_email: Optional[str] = None):
|
|
all_shares = await self._load_all_shares()
|
|
if share_id not in all_shares:
|
|
return
|
|
|
|
owner_email = all_shares[share_id]["owner_email"]
|
|
user_shares = await self._load_shares(owner_email)
|
|
|
|
if share_id in user_shares:
|
|
user_shares[share_id]["access_count"] = user_shares[share_id].get("access_count", 0) + 1
|
|
user_shares[share_id]["last_accessed"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
|
|
await self._save_shares(owner_email, user_shares)
|
|
|
|
if accessor_email:
|
|
recipients = await self._load_share_recipients(share_id)
|
|
if accessor_email in recipients:
|
|
recipients[accessor_email]["accessed"] = True
|
|
recipients[accessor_email]["last_accessed"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
|
|
await self._save_share_recipients(share_id, recipients)
|
|
|
|
async def deactivate_share(self, owner_email: str, share_id: str) -> bool:
|
|
user_shares = await self._load_shares(owner_email)
|
|
|
|
if share_id not in user_shares:
|
|
return False
|
|
|
|
user_shares[share_id]["active"] = False
|
|
await self._save_shares(owner_email, user_shares)
|
|
|
|
logger.info(f"Deactivated share {share_id}")
|
|
return True
|
|
|
|
async def reactivate_share(self, owner_email: str, share_id: str) -> bool:
|
|
user_shares = await self._load_shares(owner_email)
|
|
|
|
if share_id not in user_shares:
|
|
return False
|
|
|
|
user_shares[share_id]["active"] = True
|
|
await self._save_shares(owner_email, user_shares)
|
|
|
|
logger.info(f"Reactivated share {share_id}")
|
|
return True
|
|
|
|
async def delete_share(self, owner_email: str, share_id: str) -> bool:
|
|
user_shares = await self._load_shares(owner_email)
|
|
|
|
if share_id not in user_shares:
|
|
return False
|
|
|
|
del user_shares[share_id]
|
|
await self._save_shares(owner_email, user_shares)
|
|
|
|
all_shares = await self._load_all_shares()
|
|
if share_id in all_shares:
|
|
del all_shares[share_id]
|
|
await self._save_all_shares(all_shares)
|
|
|
|
logger.info(f"Deleted share {share_id}")
|
|
return True
|
|
|
|
async def update_share_permission(self, owner_email: str, share_id: str, permission: str) -> bool:
|
|
user_shares = await self._load_shares(owner_email)
|
|
|
|
if share_id not in user_shares:
|
|
return False
|
|
|
|
user_shares[share_id]["permission"] = permission
|
|
await self._save_shares(owner_email, user_shares)
|
|
|
|
logger.info(f"Updated permission for share {share_id} to {permission}")
|
|
return True
|
|
|
|
async def update_share_expiration(self, owner_email: str, share_id: str, expiration_days: Optional[int]) -> bool:
|
|
user_shares = await self._load_shares(owner_email)
|
|
|
|
if share_id not in user_shares:
|
|
return False
|
|
|
|
if expiration_days:
|
|
expires_at = (datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=expiration_days)).isoformat()
|
|
user_shares[share_id]["expires_at"] = expires_at
|
|
else:
|
|
user_shares[share_id]["expires_at"] = None
|
|
|
|
await self._save_shares(owner_email, user_shares)
|
|
|
|
logger.info(f"Updated expiration for share {share_id}")
|
|
return True
|
|
|
|
async def add_share_recipient(self, owner_email: str, share_id: str, recipient_email: str, permission: str) -> bool:
|
|
user_shares = await self._load_shares(owner_email)
|
|
|
|
if share_id not in user_shares:
|
|
return False
|
|
|
|
recipients = await self._load_share_recipients(share_id)
|
|
recipients[recipient_email] = {
|
|
"email": recipient_email,
|
|
"permission": permission,
|
|
"invited_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
"accessed": False
|
|
}
|
|
await self._save_share_recipients(share_id, recipients)
|
|
|
|
logger.info(f"Added recipient {recipient_email} to share {share_id}")
|
|
return True
|
|
|
|
async def remove_share_recipient(self, owner_email: str, share_id: str, recipient_email: str) -> bool:
|
|
user_shares = await self._load_shares(owner_email)
|
|
|
|
if share_id not in user_shares:
|
|
return False
|
|
|
|
recipients = await self._load_share_recipients(share_id)
|
|
if recipient_email in recipients:
|
|
del recipients[recipient_email]
|
|
await self._save_share_recipients(share_id, recipients)
|
|
logger.info(f"Removed recipient {recipient_email} from share {share_id}")
|
|
return True
|
|
|
|
return False
|
|
|
|
async def update_recipient_permission(self, owner_email: str, share_id: str, recipient_email: str, permission: str) -> bool:
|
|
user_shares = await self._load_shares(owner_email)
|
|
|
|
if share_id not in user_shares:
|
|
return False
|
|
|
|
recipients = await self._load_share_recipients(share_id)
|
|
if recipient_email in recipients:
|
|
recipients[recipient_email]["permission"] = permission
|
|
await self._save_share_recipients(share_id, recipients)
|
|
logger.info(f"Updated permission for {recipient_email} in share {share_id} to {permission}")
|
|
return True
|
|
|
|
return False
|
|
|
|
async def get_share_recipients(self, share_id: str) -> Dict:
|
|
return await self._load_share_recipients(share_id)
|
|
|
|
async def list_user_shares(self, user_email: str) -> List[Dict]:
|
|
user_shares = await self._load_shares(user_email)
|
|
return list(user_shares.values())
|
|
|
|
async def get_shares_for_item(self, user_email: str, item_path: str) -> List[Dict]:
|
|
user_shares = await self._load_shares(user_email)
|
|
item_shares = [
|
|
share for share in user_shares.values()
|
|
if share["item_path"] == item_path
|
|
]
|
|
return item_shares
|