Permission System

# File: [email protected]/public/syft.pub.yaml
rules:
  - pattern: "**/*"              # Match all files in this folder
    access:
      read: ["*"]                # Everyone can read
      write: ["[email protected]"]  # Only owner can write

terminal: false                  # Continue checking parent folders

class RuleSet(BaseModel):
    """A collection of permission rules in a folder"""
    
    rules: list[Rule] = []       # List of permission rules
    terminal: bool = False       # Stop checking parent folders?
    path: str = ""               # Path to this ruleset (runtime)
    
    @classmethod
    def load(cls, filepath: Path) -> "RuleSet":
        """Load from syft.pub.yaml file"""
        with open(filepath) as f:
            data = yaml.safe_load(f) or {}
        rs = cls.model_validate(data)
        rs.path = str(filepath.parent)
        return rs

class Rule(BaseModel):
    """A single permission rule"""
    
    pattern: str                 # Glob pattern (e.g., "**/*.csv")
    access: Access               # Access levels

class Access(BaseModel):
    """Access control lists"""
    
    admin: list[str] = []        # Full control
    write: list[str] = []        # Can create/modify files
    read: list[str] = []         # Can read files

# Everyone can read
access:
  read: ["*"]
  write: ["[email protected]"]

from syft_perm import SyftPermContext
from pathlib import Path

# Initialize context for a datasite
datasite = Path("/syftbox/[email protected]")
ctx = SyftPermContext(datasite=datasite)

# Check permissions on a file
file_perm = ctx.open("public/data.csv")

# Check if user has access
if file_perm.has_read_access("[email protected]"):
    print("Bob can read this file")

if file_perm.has_write_access("[email protected]"):
    print("Bob can write to this file")

def handle_proposed_filechange_events_message(
    self, sender_email: str, proposed_events_message: ProposedFileChangesMessage
):
    """Process incoming file changes from a peer"""
    
    # Filter to only changes sender has permission to make
    allowed_changes = [
        change
        for change in proposed_events_message.proposed_file_changes
        if self.check_write_permission(sender_email, str(change.path_in_datasite))
    ]
    
    if not allowed_changes:
        return  # Reject all changes silently
    
    # Process only allowed changes
    filtered_message = ProposedFileChangesMessage(
        sender_email=proposed_events_message.sender_email,
        proposed_file_changes=allowed_changes,
    )
    
    accepted_events_message = self.event_cache.process_proposed_events_message(
        filtered_message
    )

def _route_data_events(
    self,
    data_events: list[FileChangeEvent],
    recipients: list[str],
    events_by_recipient: dict[str, list[FileChangeEvent]],
    data_event_sent_to: dict[str, set[str]],
):
    """Route data events to recipients who have read access."""
    for event in data_events:
        path_str = str(event.path_in_datasite)
        
        # Get all recipients with read permission
        readers = self._get_readers(path_str, recipients)
        
        # Send event only to authorized readers
        for reader in readers:
            events_by_recipient[reader].append(event)

def _get_readers(self, path: str, recipients: list[str]) -> frozenset[str]:
    """Return recipients that have read access to the given path."""
    return frozenset(r for r in recipients if self.check_read_permissions(r, path))

from syft_client.job_auto_approval import (
    create_approval_policy,
    job_matches_criteria
)
from syft_job import get_client

client = get_client("/path/to/syftbox", "[email protected]")

# Define auto-approval criteria
EXPECTED_SCRIPT = """
#!/bin/bash
set -e
python count.py
"""

policy = create_approval_policy(
    required_scripts={"run.sh": EXPECTED_SCRIPT},
    required_filenames=["count.py"],
    allowed_users=["[email protected]", "[email protected]"],
    peers_only=True,  # Only approved peers
    auto_approve=True
)

# Check jobs against policy
for job in client.jobs:
    if job.status == "inbox" and job_matches_criteria(
        job,
        required_scripts=policy["required_scripts"],
        required_filenames=policy["required_filenames"],
        allowed_users=policy["allowed_users"],
        peers_only=policy["peers_only"],
    ):
        job.approve()  # Auto-approve matching jobs