""" Member Management Service for WHOOSH - Handles project member invitations, roles, and collaboration. Integrates with GITEA for repository access and Age encryption for secure communication. """ import os import json import smtplib import secrets import hashlib from pathlib import Path from typing import List, Dict, Optional, Any, Tuple from datetime import datetime, timedelta from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email import encoders from app.services.gitea_service import GiteaService from app.services.age_service import AgeService class MemberService: """ Member management service for WHOOSH project collaboration. Handles invitations, role assignments, Age key distribution, and GITEA permissions. """ def __init__(self): self.gitea_service = GiteaService() self.age_service = AgeService() self.invitations_storage = Path("/home/tony/AI/secrets/member_invitations") self.invitations_storage.mkdir(parents=True, exist_ok=True) # GITEA collaboration roles mapping self.gitea_roles = { "owner": "admin", # Full administrative access "maintainer": "write", # Write access, can merge PRs "developer": "write", # Write access, standard development "viewer": "read" # Read-only access } # Role permissions mapping self.role_permissions = { "owner": [ "repo.admin", "repo.write", "repo.read", "repo.delete", "issues.write", "issues.read", "issues.assign", "pulls.write", "pulls.read", "pulls.merge", "members.invite", "members.manage", "members.remove", "settings.configure", "age.manage" ], "maintainer": [ "repo.write", "repo.read", "issues.write", "issues.read", "issues.assign", "pulls.write", "pulls.read", "pulls.merge", "members.invite", "age.decrypt" ], "developer": [ "repo.write", "repo.read", "issues.write", "issues.read", "pulls.write", "pulls.read", "age.decrypt" ], "viewer": [ "repo.read", "issues.read", "pulls.read" ] } def generate_member_invitation(self, project_id: str, member_email: str, role: str, inviter_name: str, project_name: str, custom_message: Optional[str] = None) -> Dict[str, Any]: """ Generate a secure invitation for a project member. Args: project_id: Project identifier member_email: Email address of the invitee role: Role to assign (owner, maintainer, developer, viewer) inviter_name: Name of the person sending the invitation project_name: Human-readable project name custom_message: Optional custom message from inviter Returns: Invitation details and security tokens """ try: # Generate secure invitation token invitation_token = secrets.token_urlsafe(32) invitation_id = f"inv_{project_id}_{hashlib.sha256(member_email.encode()).hexdigest()[:8]}" # Create expiration (7 days from now) expires_at = datetime.now() + timedelta(days=7) # Create invitation record invitation_data = { "invitation_id": invitation_id, "invitation_token": invitation_token, "project_id": project_id, "project_name": project_name, "member_email": member_email, "role": role, "inviter_name": inviter_name, "custom_message": custom_message, "permissions": self.role_permissions.get(role, []), "created_at": datetime.now().isoformat(), "expires_at": expires_at.isoformat(), "status": "pending", "gitea_role": self.gitea_roles.get(role, "read"), "age_key_access": role in ["owner", "maintainer", "developer"], "responses": [], "metadata": { "invitation_method": "email", "security_level": "standard", "requires_age_key": True } } # Store invitation securely invitation_file = self.invitations_storage / f"{invitation_id}.json" invitation_file.write_text(json.dumps(invitation_data, indent=2)) invitation_file.chmod(0o600) # Restrict access print(f"Generated invitation for {member_email} to join {project_name} as {role}") print(f"Invitation ID: {invitation_id}") print(f"Expires: {expires_at.strftime('%Y-%m-%d %H:%M:%S')}") return { "invitation_id": invitation_id, "invitation_token": invitation_token, "member_email": member_email, "role": role, "expires_at": expires_at.isoformat(), "invitation_url": self._generate_invitation_url(invitation_id, invitation_token), "permissions": self.role_permissions.get(role, []), "created": True } except Exception as e: print(f"Error generating member invitation: {e}") return { "invitation_id": None, "created": False, "error": str(e) } def _generate_invitation_url(self, invitation_id: str, invitation_token: str) -> str: """Generate secure invitation URL for member to accept.""" base_url = os.getenv("WHOOSH_BASE_URL", "http://localhost:3000") return f"{base_url}/invite/{invitation_id}?token={invitation_token}" def send_email_invitation(self, invitation_data: Dict[str, Any], age_public_key: Optional[str] = None) -> bool: """ Send email invitation to project member. Args: invitation_data: Invitation details from generate_member_invitation age_public_key: Optional Age public key for encrypted communication Returns: Success status """ try: # Email configuration (using system sendmail or SMTP) smtp_config = self._get_smtp_config() if not smtp_config: print("No SMTP configuration found. Invitation email not sent.") return False # Create email content subject = f"Invitation to join {invitation_data['project_name']} on WHOOSH" # Create HTML email body email_body = self._create_invitation_email_body(invitation_data, age_public_key) # Create email message msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = smtp_config['from_email'] msg['To'] = invitation_data['member_email'] # Add HTML content html_part = MIMEText(email_body, 'html') msg.attach(html_part) # Add Age public key as attachment if provided if age_public_key: key_attachment = MIMEBase('application', 'octet-stream') key_content = f"# Age Public Key for {invitation_data['project_name']}\n{age_public_key}" key_attachment.set_payload(key_content.encode()) encoders.encode_base64(key_attachment) key_attachment.add_header( 'Content-Disposition', f'attachment; filename="{invitation_data["project_id"]}_public_key.age"' ) msg.attach(key_attachment) # Send email with smtplib.SMTP(smtp_config['smtp_host'], smtp_config['smtp_port']) as server: if smtp_config.get('use_tls'): server.starttls() if smtp_config.get('username'): server.login(smtp_config['username'], smtp_config['password']) server.send_message(msg) print(f"Invitation email sent to {invitation_data['member_email']}") # Update invitation status self._update_invitation_status( invitation_data['invitation_id'], "email_sent", {"email_sent_at": datetime.now().isoformat()} ) return True except Exception as e: print(f"Error sending invitation email: {e}") self._update_invitation_status( invitation_data['invitation_id'], "email_failed", {"email_error": str(e)} ) return False def _get_smtp_config(self) -> Optional[Dict[str, Any]]: """Get SMTP configuration from environment or secrets.""" try: # Try to load from secrets file first smtp_config_path = Path("/home/tony/AI/secrets/smtp_config.json") if smtp_config_path.exists(): return json.loads(smtp_config_path.read_text()) # Fallback to environment variables smtp_host = os.getenv("SMTP_HOST") if smtp_host: return { "smtp_host": smtp_host, "smtp_port": int(os.getenv("SMTP_PORT", "587")), "from_email": os.getenv("SMTP_FROM_EMAIL", "noreply@whoosh.local"), "username": os.getenv("SMTP_USERNAME"), "password": os.getenv("SMTP_PASSWORD"), "use_tls": os.getenv("SMTP_USE_TLS", "true").lower() == "true" } return None except Exception as e: print(f"Error loading SMTP configuration: {e}") return None def _create_invitation_email_body(self, invitation_data: Dict[str, Any], age_public_key: Optional[str] = None) -> str: """Create HTML email body for member invitation.""" # Calculate days until expiration expires_at = datetime.fromisoformat(invitation_data['expires_at']) days_until_expiry = (expires_at - datetime.now()).days role_descriptions = { "owner": "Full administrative access to the project", "maintainer": "Write access with merge permissions", "developer": "Write access for development work", "viewer": "Read-only access to project resources" } html_body = f""" WHOOSH Project Invitation

🚀 You're Invited to Join

{invitation_data['project_name']}

Hi there!

{invitation_data['inviter_name']} has invited you to collaborate on the project {invitation_data['project_name']} through the WHOOSH platform.

{"
" + invitation_data['custom_message'] + "
" if invitation_data.get('custom_message') else ""}

Your Role: {invitation_data['role'].title()}

{role_descriptions.get(invitation_data['role'], 'Custom role with specific permissions')}

🔐 Your Permissions

    {"".join(f"
  • {perm.replace('_', ' ').replace('.', ': ').title()}
    • " for perm in invitation_data['permissions'][:6])} {f"
  • ... and {len(invitation_data['permissions']) - 6} more
    • " if len(invitation_data['permissions']) > 6 else ""}
{f'''

🔒 Secure Communication

This project uses Age encryption for secure member communication. An Age public key is attached to this email for encrypted data exchange.

Once you join, you'll receive access to the project's encryption keys for secure collaboration.

''' if age_public_key else ''}

⏰ Time Sensitive: This invitation expires in {days_until_expiry} days ({expires_at.strftime('%B %d, %Y at %I:%M %p')})

Accept Invitation

What happens next?

  1. Accept the invitation using the button above
  2. Set up your WHOOSH account (if you don't have one)
  3. Gain access to the project repository and collaboration tools
  4. Start collaborating with the team immediately

🛠️ WHOOSH Features You'll Access

If you have any questions about this invitation or need help getting started, feel free to reach out to {invitation_data['inviter_name']} or the WHOOSH support team.

This invitation was sent by WHOOSH Project Management Platform

If you believe you received this invitation in error, please ignore this email.

Invitation ID: {invitation_data['invitation_id']}

""" return html_body def accept_invitation(self, invitation_id: str, invitation_token: str, accepter_data: Dict[str, Any]) -> Dict[str, Any]: """ Process invitation acceptance and set up member access. Args: invitation_id: Invitation identifier invitation_token: Security token for verification accepter_data: Data from the person accepting (name, username, etc.) Returns: Setup results and next steps """ try: # Load and validate invitation invitation = self._load_invitation(invitation_id) if not invitation: return {"success": False, "error": "Invitation not found"} if invitation["status"] != "pending": return {"success": False, "error": f"Invitation already {invitation['status']}"} if invitation["invitation_token"] != invitation_token: return {"success": False, "error": "Invalid invitation token"} # Check expiration expires_at = datetime.fromisoformat(invitation["expires_at"]) if datetime.now() > expires_at: return {"success": False, "error": "Invitation has expired"} # Extract setup data project_id = invitation["project_id"] member_email = invitation["member_email"] role = invitation["role"] gitea_role = invitation["gitea_role"] # Set up GITEA repository access gitea_setup = self._setup_gitea_member_access( project_id, member_email, gitea_role, accepter_data ) # Set up Age encryption access if required age_setup = None if invitation["age_key_access"]: age_setup = self._setup_age_member_access( project_id, member_email, role, accepter_data ) # Update invitation status self._update_invitation_status( invitation_id, "accepted", { "accepted_at": datetime.now().isoformat(), "accepter_data": accepter_data, "gitea_setup": gitea_setup, "age_setup": age_setup } ) return { "success": True, "member_email": member_email, "role": role, "project_id": project_id, "project_name": invitation["project_name"], "gitea_access": gitea_setup, "age_access": age_setup, "permissions": invitation["permissions"], "next_steps": self._generate_next_steps(invitation, gitea_setup, age_setup) } except Exception as e: print(f"Error accepting invitation: {e}") return {"success": False, "error": str(e)} def _setup_gitea_member_access(self, project_id: str, member_email: str, gitea_role: str, accepter_data: Dict[str, Any]) -> Dict[str, Any]: """Set up GITEA repository access for new member.""" try: # Get project repository info # Note: This would need to be coordinated with project service to get repo details # For now, assume standard naming convention repo_owner = "whoosh" # Default organization repo_name = project_id # Add collaborator to repository # Note: GITEA API for adding collaborators would be implemented here # For now, return setup information return { "gitea_username": accepter_data.get("gitea_username", member_email.split("@")[0]), "repository": f"{repo_owner}/{repo_name}", "role": gitea_role, "access_granted": True, "repository_url": f"{self.gitea_service.gitea_base_url}/{repo_owner}/{repo_name}" } except Exception as e: print(f"Error setting up GITEA access: {e}") return {"access_granted": False, "error": str(e)} def _setup_age_member_access(self, project_id: str, member_email: str, role: str, accepter_data: Dict[str, Any]) -> Dict[str, Any]: """Set up Age encryption access for new member.""" try: # Get project Age keys project_keys = self.age_service.list_project_keys(project_id) if not project_keys: return {"age_access": False, "error": "No Age keys found for project"} # For now, provide the public key for encrypted communication # In a full implementation, this would involve key exchange protocols primary_key = project_keys[0] return { "age_public_key": primary_key["public_key"], "key_id": primary_key["key_id"], "encryption_enabled": True, "member_can_decrypt": role in ["owner", "maintainer", "developer"], "setup_instructions": "Save the Age public key for encrypting data to this project" } except Exception as e: print(f"Error setting up Age access: {e}") return {"age_access": False, "error": str(e)} def _generate_next_steps(self, invitation: Dict, gitea_setup: Dict, age_setup: Optional[Dict]) -> List[str]: """Generate personalized next steps for new member.""" steps = [ f"Welcome to {invitation['project_name']}! Your {invitation['role']} access is now active.", ] if gitea_setup.get("access_granted"): steps.append(f"Clone the repository: git clone {gitea_setup.get('repository_url')}") steps.append("Review the project README and documentation") if age_setup and age_setup.get("encryption_enabled"): steps.append("Set up Age encryption for secure communication") if age_setup.get("member_can_decrypt"): steps.append("Contact project owner for private key access (if needed)") steps.extend([ "Check project issues and BZZZ tasks for available work", "Join the project communication channels", "Review project settings and configuration" ]) return steps def _load_invitation(self, invitation_id: str) -> Optional[Dict[str, Any]]: """Load invitation data from secure storage.""" try: invitation_file = self.invitations_storage / f"{invitation_id}.json" if invitation_file.exists(): return json.loads(invitation_file.read_text()) return None except Exception as e: print(f"Error loading invitation {invitation_id}: {e}") return None def _update_invitation_status(self, invitation_id: str, status: str, metadata: Optional[Dict[str, Any]] = None): """Update invitation status and metadata.""" try: invitation = self._load_invitation(invitation_id) if invitation: invitation["status"] = status invitation["updated_at"] = datetime.now().isoformat() if metadata: invitation.setdefault("responses", []).append({ "timestamp": datetime.now().isoformat(), "status": status, "metadata": metadata }) invitation_file = self.invitations_storage / f"{invitation_id}.json" invitation_file.write_text(json.dumps(invitation, indent=2)) except Exception as e: print(f"Error updating invitation status: {e}") def list_project_members(self, project_id: str) -> List[Dict[str, Any]]: """List all members of a project with their roles and status.""" members = [] try: # Search for all invitations related to this project for invitation_file in self.invitations_storage.glob("*.json"): try: invitation = json.loads(invitation_file.read_text()) if invitation.get("project_id") == project_id: member_info = { "email": invitation["member_email"], "role": invitation["role"], "status": invitation["status"], "invited_at": invitation["created_at"], "invited_by": invitation["inviter_name"], "permissions": invitation["permissions"] } if invitation["status"] == "accepted": # Add acceptance details for response in invitation.get("responses", []): if response.get("status") == "accepted": member_info["accepted_at"] = response["timestamp"] member_info["accepter_data"] = response.get("metadata", {}).get("accepter_data", {}) break members.append(member_info) except Exception as e: print(f"Error reading invitation file {invitation_file}: {e}") continue return members except Exception as e: print(f"Error listing project members: {e}") return [] def revoke_member_access(self, project_id: str, member_email: str, revoked_by: str, reason: str = "") -> Dict[str, Any]: """Revoke member access to a project.""" try: # Find the member's invitation for invitation_file in self.invitations_storage.glob("*.json"): try: invitation = json.loads(invitation_file.read_text()) if (invitation.get("project_id") == project_id and invitation.get("member_email") == member_email): # Update invitation status self._update_invitation_status( invitation["invitation_id"], "revoked", { "revoked_by": revoked_by, "revoke_reason": reason, "revoked_at": datetime.now().isoformat() } ) return { "success": True, "member_email": member_email, "revoked_by": revoked_by, "revoke_reason": reason } except Exception as e: print(f"Error processing invitation file {invitation_file}: {e}") continue return {"success": False, "error": "Member not found"} except Exception as e: print(f"Error revoking member access: {e}") return {"success": False, "error": str(e)} def get_invitation_status(self, invitation_id: str) -> Optional[Dict[str, Any]]: """Get current status of an invitation.""" invitation = self._load_invitation(invitation_id) if invitation: return { "invitation_id": invitation_id, "status": invitation["status"], "project_name": invitation["project_name"], "member_email": invitation["member_email"], "role": invitation["role"], "created_at": invitation["created_at"], "expires_at": invitation["expires_at"], "is_expired": datetime.now() > datetime.fromisoformat(invitation["expires_at"]) } return None def validate_invitation_token(self, invitation_id: str, token: str) -> bool: """Validate an invitation token for security.""" invitation = self._load_invitation(invitation_id) if invitation: return invitation.get("invitation_token") == token return False