Certificate Operations

def add(certificate, certificate_add_options=None, custom_headers=None, raw=False, **operation_config):
    """
    Add a certificate to the specified account.
    
    Args:
        certificate: The certificate to add (CertificateAddParameter)
        certificate_add_options: Additional options for the operation
        custom_headers: Custom headers to include in request
        raw: Return raw response if True
        
    Returns:
        None
    """

def list(certificate_list_options=None, custom_headers=None, raw=False, **operation_config):
    """
    List all certificates in the account.
    
    Args:
        certificate_list_options: Additional options for listing certificates
        
    Returns:
        ItemPaged[Certificate]: Paginated list of certificates
    """

def get(thumbprint_algorithm, thumbprint, certificate_get_options=None, custom_headers=None, raw=False, **operation_config):
    """
    Get information about the specified certificate.
    
    Args:
        thumbprint_algorithm: Algorithm used for the thumbprint (typically 'sha1')
        thumbprint: Thumbprint of the certificate
        certificate_get_options: Additional options for the operation
        
    Returns:
        Certificate: Certificate information
    """

def delete(thumbprint_algorithm, thumbprint, certificate_delete_options=None, custom_headers=None, raw=False, **operation_config):
    """
    Delete the specified certificate.
    
    Args:
        thumbprint_algorithm: Algorithm used for the thumbprint
        thumbprint: Thumbprint of the certificate to delete
        certificate_delete_options: Additional options for deletion
        
    Returns:
        None
    """

def cancel_deletion(thumbprint_algorithm, thumbprint, certificate_cancel_deletion_options=None, custom_headers=None, raw=False, **operation_config):
    """
    Cancel a failed deletion of a certificate.
    
    This can be used to cancel the deletion of a certificate if the delete operation
    encountered an error (such as being used by a pool or compute node).
    
    Args:
        thumbprint_algorithm: Algorithm used for the thumbprint
        thumbprint: Thumbprint of the certificate
        certificate_cancel_deletion_options: Additional options
        
    Returns:
        None
    """

from azure.batch.models import CertificateAddParameter
import base64

# Add a PFX certificate with password
with open("mycert.pfx", "rb") as cert_file:
    cert_data = base64.b64encode(cert_file.read()).decode('utf-8')

pfx_cert = CertificateAddParameter(
    thumbprint="abcdef1234567890abcdef1234567890abcdef12",
    thumbprint_algorithm="sha1",
    data=cert_data,
    certificate_format="pfx",
    password="certificate_password"
)

client.certificate.add(pfx_cert)

# Add a CER certificate (public key only, no password)
with open("mycert.cer", "rb") as cert_file:
    cert_data = base64.b64encode(cert_file.read()).decode('utf-8')

cer_cert = CertificateAddParameter(
    thumbprint="1234567890abcdef1234567890abcdef12345678",
    thumbprint_algorithm="sha1", 
    data=cert_data,
    certificate_format="cer"
)

client.certificate.add(cer_cert)

# List all certificates
certificates = client.certificate.list()
for cert in certificates:
    print(f"Certificate: {cert.thumbprint}")
    print(f"  State: {cert.state}")
    print(f"  Format: {cert.certificate_format}")
    print(f"  Subject: {cert.subject_name}")
    if cert.delete_certificate_error:
        print(f"  Delete Error: {cert.delete_certificate_error.message}")

# Get specific certificate details
cert = client.certificate.get("sha1", "abcdef1234567890abcdef1234567890abcdef12")
print(f"Certificate Subject: {cert.subject_name}")
print(f"Expiry Date: {cert.expiry_date}")
print(f"State: {cert.state}")

from azure.batch.models import (
    PoolSpecification, CertificateReference,
    CertificateStoreLocation, CertificateVisibility
)

# Reference certificates in pool configuration
cert_refs = [
    CertificateReference(
        thumbprint="abcdef1234567890abcdef1234567890abcdef12",
        thumbprint_algorithm="sha1",
        store_location="localmachine",  # localmachine or currentuser (Windows)
        store_name="My",  # Certificate store name
        visibility=["starttask", "task", "remoteuser"]  # Who can access the cert
    )
]

pool_spec = PoolSpecification(
    id="secure-pool",
    vm_size="Standard_D2s_v3",
    target_dedicated_nodes=2,
    certificate_references=cert_refs
)

client.pool.add(pool_spec)

# Check certificate state and handle deletion errors
cert = client.certificate.get("sha1", "abcdef1234567890abcdef1234567890abcdef12")

if cert.state == "deletefailed":
    print("Certificate deletion failed, canceling deletion")
    client.certificate.cancel_deletion("sha1", "abcdef1234567890abcdef1234567890abcdef12")
    
    # Try deleting again after a delay
    import time
    time.sleep(30)
    client.certificate.delete("sha1", "abcdef1234567890abcdef1234567890abcdef12")

elif cert.state == "active":
    print("Certificate is active and ready to use")
    
elif cert.state == "deleting":
    print("Certificate deletion in progress")

from azure.batch.models import CertificateListOptions

# List certificates with filtering
list_options = CertificateListOptions(
    filter="state eq 'active'",  # Only active certificates
    select="thumbprint,state,subjectName",  # Specific properties only
    max_results=50
)

active_certs = client.certificate.list(list_options)
for cert in active_certs:
    print(f"Active Certificate: {cert.thumbprint} - {cert.subject_name}")

# Batch certificate cleanup
def cleanup_expired_certificates():
    """Remove certificates that are expired or in error state."""
    certificates = client.certificate.list()
    
    for cert in certificates:
        try:
            if cert.state in ["deletefailed", "inactive"]:
                print(f"Attempting to delete certificate: {cert.thumbprint}")
                
                if cert.state == "deletefailed":
                    # Cancel previous failed deletion first
                    client.certificate.cancel_deletion(
                        cert.thumbprint_algorithm, cert.thumbprint
                    )
                    time.sleep(5)
                
                # Delete the certificate
                client.certificate.delete(cert.thumbprint_algorithm, cert.thumbprint)
                
        except Exception as e:
            print(f"Failed to delete certificate {cert.thumbprint}: {e}")

# Usage
cleanup_expired_certificates()

class Certificate:
    """Certificate information and state."""
    def __init__(self):
        self.thumbprint: str
        self.thumbprint_algorithm: str
        self.url: str
        self.state: str  # active, deleting, deletefailed
        self.state_transition_time: datetime.datetime
        self.public_data: str
        self.certificate_format: str  # pfx, cer
        self.provisioning_state: str
        self.provisioning_state_transition_time: datetime.datetime
        self.previous_provisioning_state: str
        self.previous_provisioning_state_transition_time: datetime.datetime
        self.subject_name: str
        self.expiry_date: datetime.datetime
        self.delete_certificate_error: DeleteCertificateError

class CertificateAddParameter:
    """Parameters for adding a certificate."""
    def __init__(self):
        self.thumbprint: str
        self.thumbprint_algorithm: str
        self.data: str  # Base64 encoded certificate data
        self.certificate_format: str  # pfx, cer
        self.password: str  # Required for PFX certificates

class CertificateReference:
    """Reference to a certificate for use in pools/tasks."""
    def __init__(self):
        self.thumbprint: str
        self.thumbprint_algorithm: str
        self.store_location: str  # localmachine, currentuser
        self.store_name: str
        self.visibility: List[str]  # starttask, task, remoteuser

class DeleteCertificateError:
    """Error information for failed certificate deletion."""
    def __init__(self):
        self.code: str
        self.message: str
        self.values: List[NameValuePair]

class CertificateListOptions:
    """Options for listing certificates."""
    def __init__(self):
        self.filter: str
        self.select: str
        self.max_results: int
        self.timeout: int

class CertificateGetOptions:
    """Options for getting certificate information.""" 
    def __init__(self):
        self.select: str
        self.timeout: int

class CertificateAddOptions:
    """Options for adding certificates."""
    def __init__(self):
        self.timeout: int

class CertificateDeleteOptions:
    """Options for deleting certificates."""
    def __init__(self):
        self.timeout: int

class CertificateCancelDeletionOptions:
    """Options for canceling certificate deletion."""
    def __init__(self):
        self.timeout: int