Google Cloud reCAPTCHA Enterprise

from google.cloud import recaptchaenterprise

from google.cloud import recaptchaenterprise_v1

from google.cloud.recaptchaenterprise import RecaptchaEnterpriseServiceClient
from google.cloud.recaptchaenterprise import RecaptchaEnterpriseServiceAsyncClient

from google.cloud import recaptchaenterprise
from google.cloud.recaptchaenterprise_v1.types import Assessment, Event

# Initialize the client
client = recaptchaenterprise.RecaptchaEnterpriseServiceClient()

# Create an assessment to analyze user interaction
project_path = f"projects/{project_id}"

# Create event details for assessment
event = Event(
    site_key=site_key,
    user_agent="Mozilla/5.0...",
    user_ip_address="192.168.1.1",
    expected_action="login"
)

# Create assessment request
assessment = Assessment(event=event)
request = recaptchaenterprise.CreateAssessmentRequest(
    parent=project_path,
    assessment=assessment
)

# Execute assessment
response = client.create_assessment(request=request)

# Analyze results
if response.risk_analysis.score > 0.5:
    print("User interaction appears legitimate")
    # Allow action
else:
    print("Potential fraud detected")
    # Block or challenge user

def create_assessment(request: CreateAssessmentRequest) -> Assessment: ...
def annotate_assessment(request: AnnotateAssessmentRequest) -> AnnotateAssessmentResponse: ...

def create_key(request: CreateKeyRequest) -> Key: ...
def list_keys(request: ListKeysRequest) -> ListKeysResponse: ...
def get_key(request: GetKeyRequest) -> Key: ...
def update_key(request: UpdateKeyRequest) -> Key: ...
def delete_key(request: DeleteKeyRequest) -> None: ...
def migrate_key(request: MigrateKeyRequest) -> Key: ...
def retrieve_legacy_secret_key(request: RetrieveLegacySecretKeyRequest) -> RetrieveLegacySecretKeyResponse: ...

def create_firewall_policy(request: CreateFirewallPolicyRequest) -> FirewallPolicy: ...
def list_firewall_policies(request: ListFirewallPoliciesRequest) -> ListFirewallPoliciesResponse: ...
def get_firewall_policy(request: GetFirewallPolicyRequest) -> FirewallPolicy: ...
def update_firewall_policy(request: UpdateFirewallPolicyRequest) -> FirewallPolicy: ...
def delete_firewall_policy(request: DeleteFirewallPolicyRequest) -> None: ...
def reorder_firewall_policies(request: ReorderFirewallPoliciesRequest) -> ReorderFirewallPoliciesResponse: ...

def add_ip_override(request: AddIpOverrideRequest) -> AddIpOverrideResponse: ...
def remove_ip_override(request: RemoveIpOverrideRequest) -> RemoveIpOverrideResponse: ...
def list_ip_overrides(request: ListIpOverridesRequest) -> ListIpOverridesResponse: ...

def list_related_account_groups(request: ListRelatedAccountGroupsRequest) -> ListRelatedAccountGroupsResponse: ...
def list_related_account_group_memberships(request: ListRelatedAccountGroupMembershipsRequest) -> ListRelatedAccountGroupMembershipsResponse: ...
def search_related_account_group_memberships(request: SearchRelatedAccountGroupMembershipsRequest) -> SearchRelatedAccountGroupMembershipsResponse: ...

def get_metrics(request: GetMetricsRequest) -> Metrics: ...

class Assessment:
    """Assessment of user interaction for fraud detection."""
    name: str
    event: Event
    risk_analysis: RiskAnalysis
    token_properties: TokenProperties
    account_defender_assessment: AccountDefenderAssessment
    fraud_prevention_assessment: FraudPreventionAssessment
    phone_fraud_assessment: PhoneFraudAssessment
    firewall_policy_assessment: FirewallPolicyAssessment

class Event:
    """User event details for assessment."""
    token: str
    site_key: str
    user_agent: str
    user_ip_address: str
    expected_action: str
    hashed_account_id: bytes
    express: bool
    requested_uri: str
    waf_token_assessment: bool
    ja3: str
    headers: List[str]
    firewall_policy_evaluation: bool

class RiskAnalysis:
    """Risk analysis results from assessment."""
    score: float
    reasons: List[str]
    extended_verdict: ExtendedVerdict

class TokenProperties:
    """Properties of the reCAPTCHA token."""
    valid: bool
    invalid_reason: str
    hostname: str
    android_package_name: str
    ios_bundle_id: str
    action: str
    create_time: Timestamp

class Key:
    """reCAPTCHA key configuration."""
    name: str
    display_name: str
    web_settings: WebKeySettings
    android_settings: AndroidKeySettings
    ios_settings: IOSKeySettings
    labels: Dict[str, str]
    create_time: Timestamp
    testing_options: TestingOptions
    waf_settings: WafSettings

class WebKeySettings:
    """Settings for web-based keys."""
    allow_all_domains: bool
    allowed_domains: List[str]
    allow_amp_traffic: bool
    integration_type: str
    challenge_security_preference: str

class AndroidKeySettings:
    """Settings for Android app keys."""
    allow_all_package_names: bool
    allowed_package_names: List[str]
    support_non_google_app_store_distribution: bool

class IOSKeySettings:
    """Settings for iOS app keys."""
    allow_all_bundle_ids: bool
    allowed_bundle_ids: List[str]
    apple_developer_id: AppleDeveloperId

class RecaptchaEnterpriseServiceClient:
    """Synchronous client for reCAPTCHA Enterprise API operations."""
    def __init__(self, *, credentials=None, transport=None, client_options=None, client_info=None): ...
    def create_assessment(self, request=None, **kwargs) -> Assessment: ...
    def annotate_assessment(self, request=None, **kwargs) -> AnnotateAssessmentResponse: ...
    # ... all other synchronous API methods

class RecaptchaEnterpriseServiceAsyncClient:
    """Asynchronous client for reCAPTCHA Enterprise API operations."""  
    def __init__(self, *, credentials=None, transport=None, client_options=None, client_info=None): ...
    async def create_assessment(self, request=None, **kwargs) -> Assessment: ...
    async def annotate_assessment(self, request=None, **kwargs) -> AnnotateAssessmentResponse: ...
    # ... all other asynchronous API methods

class FirewallPolicy:
    """Firewall policy configuration."""
    name: str
    description: str
    path: str
    condition: str
    actions: List[FirewallAction]

class FirewallAction:
    """Action to take when firewall conditions match."""
    allow: AllowAction
    block: BlockAction
    substitute: SubstituteAction
    redirect: RedirectAction
    set_header: SetHeaderAction

class Timestamp:
    """Google protobuf timestamp type for representing time."""
    seconds: int      # Seconds since Unix epoch
    nanos: int        # Nanoseconds within the second

class FieldMask:
    """Google protobuf field mask for specifying fields to update."""
    paths: List[str]  # Field paths in dot notation (e.g., "display_name", "web_settings.allowed_domains")

# Using Application Default Credentials
client = recaptchaenterprise.RecaptchaEnterpriseServiceClient()

# Using service account key file
from google.oauth2 import service_account
credentials = service_account.Credentials.from_service_account_file(
    "/path/to/service-account-key.json"
)
client = recaptchaenterprise.RecaptchaEnterpriseServiceClient(credentials=credentials)

# Custom endpoint configuration
client = recaptchaenterprise.RecaptchaEnterpriseServiceClient(
    client_options={"api_endpoint": "https://custom-endpoint.googleapis.com"}
)

from google.api_core import exceptions

try:
    response = client.create_assessment(request=request)
except exceptions.GoogleAPICallError as e:
    print(f"API call failed: {e}")
except exceptions.InvalidArgument as e:
    print(f"Invalid request parameters: {e}")
except exceptions.PermissionDenied as e:
    print(f"Insufficient permissions: {e}")