SAS Token Generation - Shared Access Signatures

from azure.storage.fileshare import (
    generate_account_sas,
    generate_share_sas,
    generate_file_sas,
    ShareSasPermissions,
    FileSasPermissions,
    AccountSasPermissions,
    ResourceTypes,
    Services
)
from datetime import datetime, timedelta
from typing import Union, Optional, Callable, Any

def generate_account_sas(
    account_name: str,
    account_key: str,
    resource_types: Union[ResourceTypes, str],
    permission: Union[AccountSasPermissions, str],
    expiry: Union[datetime, str],
    start: Optional[Union[datetime, str]] = None,
    ip: Optional[str] = None,
    *,
    services: Union[Services, str] = Services(fileshare=True),
    protocol: Optional[str] = None,
    sts_hook: Optional[Callable[[str], None]] = None,
    **kwargs: Any
) -> str:
    """
    Generates a shared access signature for the file service at account level.
    
    Parameters:
        account_name: The storage account name used to generate the SAS
        account_key: The account key (access key) to generate the SAS
        resource_types: Specifies resource types accessible with the SAS
        permission: The permissions associated with the SAS
        expiry: The time at which the SAS becomes invalid
        start: The time at which the SAS becomes valid (optional)
        ip: IP address or range of IP addresses to accept requests from (optional)
        services: Specifies services the SAS token can be used with (default: fileshare only)
        protocol: Specifies protocol permitted for requests ("https" or "https,http")
        sts_hook: Debugging hook for Security Token Service (optional)
        
    Returns:
        str: The SAS token string that can be appended to resource URLs
        
    Example:
        sas_token = generate_account_sas(
            account_name="myaccount",
            account_key="mykey",
            resource_types=ResourceTypes(service=True, container=True, object=True),
            permission=AccountSasPermissions(read=True, write=True, list=True),
            expiry=datetime.utcnow() + timedelta(hours=1)
        )
    """

def generate_share_sas(
    account_name: str,
    share_name: str,
    account_key: str,
    permission: Optional[Union[ShareSasPermissions, str]] = None,
    expiry: Optional[Union[datetime, str]] = None,
    start: Optional[Union[datetime, str]] = None,
    policy_id: Optional[str] = None,
    ip: Optional[str] = None,
    *,
    protocol: Optional[str] = None,
    cache_control: Optional[str] = None,
    content_disposition: Optional[str] = None,
    content_encoding: Optional[str] = None,
    content_language: Optional[str] = None,
    content_type: Optional[str] = None,
    sts_hook: Optional[Callable[[str], None]] = None,
    **kwargs: Any
) -> str:
    """
    Generates a shared access signature for a share.
    
    Parameters:
        account_name: The storage account name used to generate the SAS
        share_name: The name of the share
        account_key: The account key to generate the SAS
        permission: The permissions associated with the SAS (optional if policy_id provided)
        expiry: The time at which the SAS becomes invalid (optional if policy_id provided)
        start: The time at which the SAS becomes valid (optional)
        policy_id: A unique value that correlates to a stored access policy (optional)
        ip: IP address or range of IP addresses to accept requests from (optional)
        protocol: Specifies protocol permitted for requests ("https" or "https,http")
        cache_control: Response header value for Cache-Control when resource accessed via SAS
        content_disposition: Response header value for Content-Disposition when accessed via SAS
        content_encoding: Response header value for Content-Encoding when accessed via SAS
        content_language: Response header value for Content-Language when accessed via SAS
        content_type: Response header value for Content-Type when accessed via SAS
        sts_hook: Debugging hook for Security Token Service (optional)
        
    Returns:
        str: The SAS token string that can be appended to share URLs
    """

def generate_file_sas(
    account_name: str,
    share_name: str,
    file_path: str,
    account_key: str,
    permission: Optional[Union[FileSasPermissions, str]] = None,
    expiry: Optional[Union[datetime, str]] = None,
    start: Optional[Union[datetime, str]] = None,
    policy_id: Optional[str] = None,
    ip: Optional[str] = None,
    *,
    protocol: Optional[str] = None,
    cache_control: Optional[str] = None,
    content_disposition: Optional[str] = None,
    content_encoding: Optional[str] = None,
    content_language: Optional[str] = None,
    content_type: Optional[str] = None,
    sts_hook: Optional[Callable[[str], None]] = None,
    **kwargs: Any
) -> str:
    """
    Generates a shared access signature for a file.
    
    Parameters:
        account_name: The storage account name used to generate the SAS
        share_name: The name of the share containing the file
        file_path: The path to the file with which to associate the SAS
        account_key: The account key to generate the SAS
        permission: The permissions associated with the SAS (optional if policy_id provided)
        expiry: The time at which the SAS becomes invalid (optional if policy_id provided)
        start: The time at which the SAS becomes valid (optional)
        policy_id: A unique value that correlates to a stored access policy (optional)
        ip: IP address or range of IP addresses to accept requests from (optional)
        protocol: Specifies protocol permitted for requests ("https" or "https,http")
        cache_control: Response header value for Cache-Control when resource accessed via SAS
        content_disposition: Response header value for Content-Disposition when accessed via SAS
        content_encoding: Response header value for Content-Encoding when accessed via SAS
        content_language: Response header value for Content-Language when accessed via SAS
        content_type: Response header value for Content-Type when accessed via SAS
        sts_hook: Debugging hook for Security Token Service (optional)
        
    Returns:
        str: The SAS token string that can be appended to file URLs
    """

class AccountSasPermissions:
    def __init__(
        self,
        read: bool = False,
        write: bool = False,
        delete: bool = False,
        list: bool = False,
        add: bool = False,
        create: bool = False,
        update: bool = False,
        process: bool = False
    ) -> None:
        """
        Account-level SAS permissions.
        
        Parameters:
            read: Valid for all signed resources types (Service, Container, Object)
            write: Valid for all signed resources types (Service, Container, Object)  
            delete: Valid for Container and Object resource types
            list: Valid for Service and Container resource types only
            add: Valid for Object resource types only (queue messages, table entities, append blobs)
            create: Valid for Object resource types only (blobs and files)
            update: Valid for Object resource types only (queue messages and table entities)
            process: Valid for Object resource types only (queue messages)
        """
    
    @classmethod
    def from_string(cls, permission: str) -> AccountSasPermissions:
        """
        Create AccountSasPermissions from a string.
        
        Parameters:
            permission: Permission string using letters (r=read, w=write, d=delete, l=list, a=add, c=create, u=update, p=process)
        """

class ShareSasPermissions:
    def __init__(
        self,
        read: bool = False,
        create: bool = False,
        write: bool = False,
        delete: bool = False,
        list: bool = False
    ) -> None:
        """
        Share-level SAS permissions.
        
        Parameters:
            read: Read the content, properties, metadata of any file in the share
            create: Create a new file in the share, or copy a file to a new file in the share
            write: Create or write content, properties, metadata. Resize the file
            delete: Delete any file in the share
            list: List files and directories in the share
        """
    
    @classmethod
    def from_string(cls, permission: str) -> ShareSasPermissions:
        """
        Create ShareSasPermissions from a string.
        
        Parameters:
            permission: Permission string using letters (r=read, c=create, w=write, d=delete, l=list)
        """

class FileSasPermissions:
    def __init__(
        self,
        read: bool = False,
        create: bool = False,
        write: bool = False,
        delete: bool = False
    ) -> None:
        """
        File-level SAS permissions.
        
        Parameters:
            read: Read the content, properties, metadata. Use file as source of copy operation
            create: Create a new file or copy a file to a new file
            write: Create or write content, properties, metadata. Resize the file
            delete: Delete the file
        """
    
    @classmethod
    def from_string(cls, permission: str) -> FileSasPermissions:
        """
        Create FileSasPermissions from a string.
        
        Parameters:
            permission: Permission string using letters (r=read, c=create, w=write, d=delete)
        """

class ResourceTypes:
    def __init__(
        self,
        service: bool = False,
        container: bool = False,
        object: bool = False
    ) -> None:
        """
        Specifies resource types accessible with the account SAS.
        
        Parameters:
            service: Access to service-level APIs (Get/Set Service Properties, List Shares)
            container: Access to container-level APIs (Create/Delete Share, Share Properties, List Directories/Files)
            object: Access to object-level APIs (Create/Delete/Read/Write File, File Properties)
        """
    
    @classmethod
    def from_string(cls, resource_types: str) -> ResourceTypes:
        """
        Create ResourceTypes from a string.
        
        Parameters:
            resource_types: String using letters (s=service, c=container, o=object)
        """

class Services:
    def __init__(
        self,
        blob: bool = False,
        queue: bool = False,
        table: bool = False,
        fileshare: bool = False
    ) -> None:
        """
        Specifies services accessible with the account SAS.
        
        Parameters:
            blob: Valid for Blob service
            queue: Valid for Queue service
            table: Valid for Table service
            fileshare: Valid for File service
        """
    
    @classmethod
    def from_string(cls, services: str) -> Services:
        """
        Create Services from a string.
        
        Parameters:
            services: String using letters (b=blob, q=queue, t=table, f=fileshare)
        """

from azure.storage.fileshare import *
from datetime import datetime, timedelta

# Account credentials
account_name = "mystorageaccount"
account_key = "myaccountkey"

# Generate account-level SAS for full file service access
account_sas = generate_account_sas(
    account_name=account_name,
    account_key=account_key,
    resource_types=ResourceTypes(service=True, container=True, object=True),
    permission=AccountSasPermissions(read=True, write=True, list=True, create=True, delete=True),
    expiry=datetime.utcnow() + timedelta(hours=24),  # Valid for 24 hours
    start=datetime.utcnow() - timedelta(minutes=5),  # Valid from 5 minutes ago
    protocol="https"  # HTTPS only
)
print(f"Account SAS: {account_sas}")

# Generate share-level SAS for specific share
share_sas = generate_share_sas(
    account_name=account_name,
    share_name="documents",
    account_key=account_key,
    permission=ShareSasPermissions(read=True, list=True),
    expiry=datetime.utcnow() + timedelta(hours=2)  # 2-hour read access
)
print(f"Share SAS: {share_sas}")

# Generate file-level SAS for specific file
file_sas = generate_file_sas(
    account_name=account_name,
    share_name="documents", 
    file_path="reports/monthly_report.pdf",
    account_key=account_key,
    permission=FileSasPermissions(read=True),
    expiry=datetime.utcnow() + timedelta(hours=1),  # 1-hour read access
    content_disposition="attachment; filename=report.pdf"  # Force download
)
print(f"File SAS: {file_sas}")

from azure.storage.fileshare import ShareServiceClient, ShareClient, ShareFileClient

# Use account SAS with service client
service_client = ShareServiceClient(
    account_url=f"https://{account_name}.file.core.windows.net",
    credential=account_sas
)

# List shares using account SAS
shares = list(service_client.list_shares())
print(f"Found {len(shares)} shares")

# Use share SAS with share client  
share_client = ShareClient(
    account_url=f"https://{account_name}.file.core.windows.net",
    share_name="documents",
    credential=share_sas
)

# List files using share SAS
files = list(share_client.list_directories_and_files())
print(f"Found {len(files)} items in share")

# Use file SAS with file client
file_client = ShareFileClient(
    account_url=f"https://{account_name}.file.core.windows.net",
    share_name="documents",
    file_path="reports/monthly_report.pdf",
    credential=file_sas
)

# Download file using file SAS
download_stream = file_client.download_file()
content = download_stream.readall()
print(f"Downloaded {len(content)} bytes")

# SAS with IP restrictions
restricted_sas = generate_file_sas(
    account_name=account_name,
    share_name="sensitive-data",
    file_path="confidential.txt",
    account_key=account_key,
    permission=FileSasPermissions(read=True),
    expiry=datetime.utcnow() + timedelta(minutes=30),
    ip="192.168.1.100",  # Only allow from specific IP
    protocol="https"     # HTTPS only
)

# SAS with custom response headers
download_sas = generate_file_sas(
    account_name=account_name,
    share_name="downloads",
    file_path="software/installer.exe",
    account_key=account_key,
    permission=FileSasPermissions(read=True),
    expiry=datetime.utcnow() + timedelta(hours=6),
    content_type="application/octet-stream",
    content_disposition="attachment; filename=installer.exe",
    cache_control="no-cache"
)

# SAS for temporary upload access
upload_sas = generate_share_sas(
    account_name=account_name,
    share_name="uploads",
    account_key=account_key,
    permission=ShareSasPermissions(create=True, write=True),
    expiry=datetime.utcnow() + timedelta(hours=1),  # 1-hour upload window
    start=datetime.utcnow()  # Valid immediately
)

print(f"Upload SAS: {upload_sas}")

from azure.storage.fileshare import AccessPolicy

# First, create a stored access policy on the share
share_client = ShareClient(
    account_url=f"https://{account_name}.file.core.windows.net",
    share_name="documents",
    credential=AzureNamedKeyCredential(account_name, account_key)
)

# Define access policy
policy = AccessPolicy(
    permission=ShareSasPermissions(read=True, list=True),
    expiry=datetime.utcnow() + timedelta(days=30),
    start=datetime.utcnow()
)

# Set the access policy on the share
share_client.set_share_access_policy({"readonly": policy})

# Generate SAS using stored access policy (no expiry/permission needed)
policy_sas = generate_share_sas(
    account_name=account_name,
    share_name="documents",
    account_key=account_key,
    policy_id="readonly"  # Reference to stored policy
)

# This SAS inherits permissions and expiry from the stored policy
print(f"Policy-based SAS: {policy_sas}")

# Benefits: Can modify policy without regenerating SAS tokens
# Update the policy to add write permissions
updated_policy = AccessPolicy(
    permission=ShareSasPermissions(read=True, list=True, write=True),
    expiry=datetime.utcnow() + timedelta(days=30),
    start=datetime.utcnow()
)

share_client.set_share_access_policy({"readonly": updated_policy})
# Existing SAS tokens now have write permissions too!

# 1. Read-only access for external users
public_read_sas = generate_share_sas(
    account_name=account_name,
    share_name="public-documents",
    account_key=account_key,
    permission=ShareSasPermissions(read=True, list=True),
    expiry=datetime.utcnow() + timedelta(days=7)  # Week-long access
)

# 2. Temporary upload for file submissions
upload_window_sas = generate_share_sas(
    account_name=account_name,
    share_name="submissions",
    account_key=account_key,
    permission=ShareSasPermissions(create=True, write=True),
    expiry=datetime.utcnow() + timedelta(hours=2),  # 2-hour upload window
    start=datetime.utcnow()
)

# 3. Backup service access
backup_sas = generate_account_sas(
    account_name=account_name,
    account_key=account_key,
    resource_types=ResourceTypes(container=True, object=True),
    permission=AccountSasPermissions(read=True, list=True),
    expiry=datetime.utcnow() + timedelta(hours=8),  # 8-hour backup window
    services=Services(fileshare=True)
)

# 4. Monitoring/audit read access
audit_sas = generate_account_sas(
    account_name=account_name,
    account_key=account_key,
    resource_types=ResourceTypes(service=True, container=True),
    permission=AccountSasPermissions(read=True, list=True),
    expiry=datetime.utcnow() + timedelta(days=1)
)

# Construct full URLs with SAS tokens
base_url = f"https://{account_name}.file.core.windows.net"

# Share URL with SAS
share_url_with_sas = f"{base_url}/documents?{share_sas}"
print(f"Share URL: {share_url_with_sas}")

# File URL with SAS  
file_url_with_sas = f"{base_url}/documents/reports/monthly_report.pdf?{file_sas}"
print(f"File URL: {file_url_with_sas}")

# Use URLs directly with HTTP clients
import requests

# Download file using direct HTTP request
response = requests.get(file_url_with_sas)
if response.status_code == 200:
    with open("downloaded_report.pdf", "wb") as f:
        f.write(response.content)
    print("File downloaded successfully")
else:
    print(f"Download failed: {response.status_code}")

# Upload file using direct HTTP request
upload_url = f"{base_url}/uploads/new_file.txt?{upload_sas}"
with open("local_file.txt", "rb") as f:
    response = requests.put(
        upload_url,
        data=f,
        headers={
            "x-ms-type": "file",
            "x-ms-content-length": str(os.path.getsize("local_file.txt"))
        }
    )
    
if response.status_code in [201, 200]:
    print("File uploaded successfully")
else:
    print(f"Upload failed: {response.status_code}")

# 1. Principle of least privilege - minimal permissions
minimal_sas = generate_file_sas(
    account_name=account_name,
    share_name="documents",
    file_path="public/readme.txt",
    account_key=account_key,
    permission=FileSasPermissions(read=True),  # Only read, no write/delete
    expiry=datetime.utcnow() + timedelta(hours=1)  # Short expiration
)

# 2. HTTPS only for sensitive data
secure_sas = generate_file_sas(
    account_name=account_name,
    share_name="confidential",
    file_path="sensitive_data.txt",
    account_key=account_key,
    permission=FileSasPermissions(read=True),
    expiry=datetime.utcnow() + timedelta(minutes=15),  # Very short expiration
    protocol="https",  # Force HTTPS
    ip="203.0.113.0/24"  # Restrict to known IP range
)

# 3. Use stored access policies for revocation capability
def create_revocable_sas(share_client, policy_name, permissions, duration):
    """Create SAS that can be revoked by updating stored policy."""
    
    # Create policy
    policy = AccessPolicy(
        permission=permissions,
        expiry=datetime.utcnow() + duration,
        start=datetime.utcnow()
    )
    
    # Set policy on share
    share_client.set_share_access_policy({policy_name: policy})
    
    # Generate SAS using policy
    return generate_share_sas(
        account_name=share_client.account_name,
        share_name=share_client.share_name,
        account_key=account_key,
        policy_id=policy_name
    )

# To revoke, just delete the policy
def revoke_sas(share_client, policy_name):
    """Revoke SAS by removing stored access policy."""
    share_client.set_share_access_policy({})  # Empty policy removes all

# 4. Monitor SAS usage (implement logging)
def generate_tracked_sas(purpose, **sas_args):
    """Generate SAS with usage tracking."""
    sas_token = generate_file_sas(**sas_args)
    
    # Log SAS generation
    print(f"Generated SAS for {purpose}")
    print(f"Expiry: {sas_args['expiry']}")
    print(f"Permissions: {sas_args['permission']}")
    
    return sas_token