Security Skills Atlas
api security

Implementing API Threat Protection with Apigee

Implement API threat protection using Google Apigee policies including JSON/XML threat protection, OAuth 2.0, SpikeArrest, and Advanced API Security for OWASP Top 10 defense.

api-gatewayapigeegoogle-cloudjson-threat-protectionoauth2owasp-api-top-10spike-arrestthreat-protection
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1059.007 on the official MITRE ATT&CK siteT1078.004 on the official MITRE ATT&CK siteT1190 on the official MITRE ATT&CK siteT1530 on the official MITRE ATT&CK siteT1552.001 on the official MITRE ATT&CK site

NIST CSF 2.0

DE.CM-01 on the official NIST CSF 2.0 siteID.RA-01 on the official NIST CSF 2.0 sitePR.DS-10 on the official NIST CSF 2.0 sitePR.PS-01 on the official NIST CSF 2.0 site

Overview

Google Apigee is an enterprise API management platform that provides native security policies for threat protection, including JSON and XML content validation, OAuth 2.0 enforcement, SpikeArrest rate limiting, regular expression threat protection, and Advanced API Security for detecting malicious clients and API abuse patterns. Apigee operates as a reverse proxy that intercepts all API traffic, applying security policies before requests reach backend services, effectively shielding APIs against the OWASP API Security Top 10 threats.

When to Use

  • When deploying or configuring implementing api threat protection with apigee capabilities in your environment
  • When establishing security controls aligned to compliance requirements
  • When building or improving security architecture for this domain
  • When conducting security assessments that require this implementation

Prerequisites

  • Google Cloud Platform account with Apigee organization provisioned
  • Apigee X or Apigee hybrid environment configured
  • Backend API services deployed and accessible from Apigee
  • Google Cloud CLI (gcloud) installed and authenticated
  • OpenAPI specification for target APIs
  • Understanding of Apigee proxy bundle structure

Core Security Policies

1. JSON Threat Protection

Protects against JSON-based denial-of-service attacks by limiting structural depth, entry counts, and string lengths:

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<JSONThreatProtection name="JSON-Threat-Protection-1">
    <DisplayName>JSON Threat Protection</DisplayName>
    <Source>request</Source>
    <!-- Maximum nesting depth of JSON structure -->
    <ObjectEntryNameLength>50</ObjectEntryNameLength>
    <ObjectEntryCount>25</ObjectEntryCount>
    <ArrayElementCount>100</ArrayElementCount>
    <ContainerDepth>5</ContainerDepth>
    <StringValueLength>500</StringValueLength>
</JSONThreatProtection>

2. XML Threat Protection

Shields against XML bombs, XXE attacks, and oversized XML payloads:

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<XMLThreatProtection name="XML-Threat-Protection-1">
    <DisplayName>XML Threat Protection</DisplayName>
    <Source>request</Source>
    <NameLimits>
        <Element>50</Element>
        <Attribute>50</Attribute>
        <NamespacePrefix>20</NamespacePrefix>
        <ProcessingInstructionTarget>50</ProcessingInstructionTarget>
    </NameLimits>
    <ValueLimits>
        <Text>1000</Text>
        <Attribute>500</Attribute>
        <NamespaceURI>256</NamespaceURI>
        <Comment>256</Comment>
        <ProcessingInstructionData>256</ProcessingInstructionData>
    </ValueLimits>
    <StructureLimits>
        <NodeDepth>5</NodeDepth>
        <AttributeCountPerElement>5</AttributeCountPerElement>
        <NamespaceCountPerElement>3</NamespaceCountPerElement>
        <ChildCount>25</ChildCount>
    </StructureLimits>
</XMLThreatProtection>

3. Regular Expression Threat Protection

Detects SQL injection, XSS, and other injection patterns in request parameters:

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<RegularExpressionProtection name="RegEx-Threat-Protection-1">
    <DisplayName>Regex Injection Protection</DisplayName>
    <Source>request</Source>
    <IgnoreUnresolvedVariables>false</IgnoreUnresolvedVariables>
 
    <!-- SQL Injection patterns -->
    <QueryParam name="*">
        <Pattern>[\s]*((delete)|(exec)|(drop\s*table)|(insert)|(shutdown)|(update)|(\bor\b))</Pattern>
    </QueryParam>
 
    <!-- XSS patterns -->
    <QueryParam name="*">
        <Pattern>[\s]*&lt;\s*script\b[^&gt;]*&gt;[^&lt;]+&lt;\s*/\s*script\s*&gt;</Pattern>
    </QueryParam>
 
    <!-- Header injection -->
    <Header name="*">
        <Pattern>[\r\n]</Pattern>
    </Header>
 
    <!-- URI path traversal -->
    <URIPath>
        <Pattern>(/\.\.)|(\.\./)</Pattern>
    </URIPath>
 
    <!-- JSON body injection -->
    <JSONPayload>
        <JSONPath>$.*</JSONPath>
        <Pattern>[\s]*((delete)|(exec)|(drop\s*table)|(insert)|(shutdown)|(update))</Pattern>
    </JSONPayload>
</RegularExpressionProtection>

4. SpikeArrest Policy

Prevents traffic spikes from overwhelming backend services:

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<SpikeArrest name="Spike-Arrest-1">
    <DisplayName>API Spike Arrest</DisplayName>
    <Rate>30ps</Rate> <!-- 30 per second smoothed -->
    <Identifier ref="request.header.x-api-key"/>
    <MessageWeight ref="request.header.x-request-weight"/>
    <UseEffectiveCount>true</UseEffectiveCount>
</SpikeArrest>

5. OAuth 2.0 Token Validation

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<OAuthV2 name="Verify-OAuth-Token">
    <DisplayName>Verify OAuth 2.0 Access Token</DisplayName>
    <Operation>VerifyAccessToken</Operation>
    <ExternalAuthorization>false</ExternalAuthorization>
    <ExternalAccessToken>request.header.Authorization</ExternalAccessToken>
    <SupportedGrantTypes>
        <GrantType>authorization_code</GrantType>
        <GrantType>client_credentials</GrantType>
    </SupportedGrantTypes>
    <Scope>read write</Scope>
    <GenerateResponse enabled="true"/>
</OAuthV2>

6. API Key Validation

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<VerifyAPIKey name="Verify-API-Key-1">
    <DisplayName>Verify API Key</DisplayName>
    <APIKey ref="request.header.x-api-key"/>
</VerifyAPIKey>

Proxy Bundle Configuration

Complete Secure Proxy Flow

<!-- apiproxy/proxies/default.xml -->
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<ProxyEndpoint name="default">
    <PreFlow name="PreFlow">
        <Request>
            <!-- Step 1: Verify API Key or OAuth token -->
            <Step>
                <Name>Verify-OAuth-Token</Name>
            </Step>
            <!-- Step 2: Rate limiting -->
            <Step>
                <Name>Spike-Arrest-1</Name>
            </Step>
            <!-- Step 3: Threat protection -->
            <Step>
                <Name>JSON-Threat-Protection-1</Name>
                <Condition>request.header.Content-Type = "application/json"</Condition>
            </Step>
            <Step>
                <Name>XML-Threat-Protection-1</Name>
                <Condition>request.header.Content-Type = "text/xml"</Condition>
            </Step>
            <!-- Step 4: Injection prevention -->
            <Step>
                <Name>RegEx-Threat-Protection-1</Name>
            </Step>
            <!-- Step 5: CORS enforcement -->
            <Step>
                <Name>CORS-Policy</Name>
            </Step>
        </Request>
        <Response>
            <!-- Remove internal headers from response -->
            <Step>
                <Name>Remove-Internal-Headers</Name>
            </Step>
            <!-- Add security headers -->
            <Step>
                <Name>Add-Security-Headers</Name>
            </Step>
        </Response>
    </PreFlow>
 
    <Flows>
        <Flow name="sensitive-operations">
            <Description>Additional protection for sensitive endpoints</Description>
            <Request>
                <Step>
                    <Name>Quota-Strict</Name>
                </Step>
            </Request>
            <Condition>(proxy.pathsuffix MatchesPath "/admin/**") or
                       (proxy.pathsuffix MatchesPath "/users/*/sensitive")</Condition>
        </Flow>
    </Flows>
 
    <HTTPProxyConnection>
        <BasePath>/v1</BasePath>
        <VirtualHost>secure</VirtualHost>
    </HTTPProxyConnection>
 
    <RouteRule name="default">
        <TargetEndpoint>default</TargetEndpoint>
    </RouteRule>
</ProxyEndpoint>

Security Headers Policy

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<AssignMessage name="Add-Security-Headers">
    <DisplayName>Add Security Response Headers</DisplayName>
    <Set>
        <Headers>
            <Header name="X-Content-Type-Options">nosniff</Header>
            <Header name="X-Frame-Options">DENY</Header>
            <Header name="Strict-Transport-Security">max-age=31536000; includeSubDomains</Header>
            <Header name="Cache-Control">no-store, no-cache, must-revalidate</Header>
            <Header name="Content-Security-Policy">default-src 'none'</Header>
            <Header name="X-Request-ID">{messageid}</Header>
        </Headers>
    </Set>
    <Remove>
        <Headers>
            <Header name="X-Powered-By"/>
            <Header name="Server"/>
        </Headers>
    </Remove>
    <IgnoreUnresolvedVariables>false</IgnoreUnresolvedVariables>
    <AssignTo createNew="false" transport="http" type="response"/>
</AssignMessage>

Advanced API Security

Enable Apigee's Advanced API Security add-on for machine-learning-based threat detection:

# Enable Advanced API Security on Apigee X instance
gcloud apigee organizations update $ORG_NAME \
  --advanced-api-security-config=enabled
 
# View detected abuse alerts
gcloud apigee apis security-reports list \
  --organization=$ORG_NAME \
  --environment=$ENV_NAME
 
# Create security action to block suspicious traffic
gcloud apigee security-actions create \
  --organization=$ORG_NAME \
  --environment=$ENV_NAME \
  --action-type=DENY \
  --condition-type=IP_ADDRESS \
  --condition-values="192.168.1.100,10.0.0.50" \
  --description="Block identified malicious IPs"

Deployment

# Deploy proxy bundle with security policies
gcloud apigee apis deploy \
  --api=$API_NAME \
  --environment=$ENV_NAME \
  --revision=$REVISION \
  --organization=$ORG_NAME
 
# Validate deployment
gcloud apigee apis list-deployments \
  --api=$API_NAME \
  --organization=$ORG_NAME

References

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md1.6 KB

API Reference: Implementing API Threat Protection with Apigee

JSONThreatProtection Policy

<JSONThreatProtection name="JSON-Threat">
    <ObjectEntryCount>25</ObjectEntryCount>
    <ArrayElementCount>100</ArrayElementCount>
    <ContainerDepth>5</ContainerDepth>
    <StringValueLength>500</StringValueLength>
    <Source>request</Source>
</JSONThreatProtection>

SpikeArrest Policy

<SpikeArrest name="Spike-Arrest">
    <Rate>30ps</Rate>
    <Identifier ref="request.header.x-api-key"/>
</SpikeArrest>

RegularExpressionProtection

<RegularExpressionProtection name="Regex-Protect">
    <Source>request</Source>
    <QueryParam name="*">
        <Pattern>[\s]*((delete)|(exec)|(drop\s*table))</Pattern>
    </QueryParam>
</RegularExpressionProtection>

Apigee Management API

# Deploy proxy revision
curl -X POST "https://apigee.googleapis.com/v1/organizations/{org}/environments/{env}/apis/{api}/revisions/{rev}/deployments" \
  -H "Authorization: Bearer $(gcloud auth print-access-token)"
 
# List deployed proxies
curl "https://apigee.googleapis.com/v1/organizations/{org}/apis" \
  -H "Authorization: Bearer $(gcloud auth print-access-token)"

Recommended Policy Limits

Setting Recommended Description
ContainerDepth 5 JSON nesting depth
StringValueLength 500 Max string value
ObjectEntryCount 25 Max object keys
SpikeArrest Rate 30ps Requests per second

References

Scripts 1

agent.py6.5 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Agent for implementing API threat protection policies with Google Apigee."""

import json
import argparse
import re
from datetime import datetime
from pathlib import Path


APIGEE_POLICIES = {
    "JSONThreatProtection": {
        "max_depth": 5, "max_string_length": 500,
        "max_entries": 25, "max_array_elements": 100,
    },
    "XMLThreatProtection": {
        "max_depth": 5, "max_attributes": 10,
        "max_element_name_length": 128, "max_text_length": 500,
    },
    "RegularExpressionProtection": {
        "patterns": [
            r"[\s]*((delete)|(exec)|(drop\s*table)|(insert)|(shutdown)|(update)|(or))",
            r"<\s*script\b[^>]*>[^<]+<\s*/\s*script\s*>",
            r"(\%27)|(\')|(\-\-)|(\%23)|(#)",
        ]
    },
    "SpikeArrest": {
        "rate": "30ps", "identifier": "request.header.x-api-key",
    },
}


def generate_json_threat_policy(config=None):
    """Generate Apigee JSONThreatProtection policy XML."""
    cfg = config or APIGEE_POLICIES["JSONThreatProtection"]
    return f"""<?xml version="1.0" encoding="UTF-8"?>
<JSONThreatProtection name="JSON-Threat-Protection">
    <DisplayName>JSON Threat Protection</DisplayName>
    <ObjectEntryCount>{cfg['max_entries']}</ObjectEntryCount>
    <ArrayElementCount>{cfg['max_array_elements']}</ArrayElementCount>
    <ContainerDepth>{cfg['max_depth']}</ContainerDepth>
    <StringValueLength>{cfg['max_string_length']}</StringValueLength>
    <Source>request</Source>
</JSONThreatProtection>"""


def generate_spike_arrest_policy(rate="30ps"):
    """Generate Apigee SpikeArrest policy XML."""
    return f"""<?xml version="1.0" encoding="UTF-8"?>
<SpikeArrest name="Spike-Arrest">
    <DisplayName>Spike Arrest</DisplayName>
    <Rate>{rate}</Rate>
    <Identifier ref="request.header.x-api-key"/>
    <UseEffectiveCount>true</UseEffectiveCount>
</SpikeArrest>"""


def generate_regex_protection_policy():
    """Generate RegularExpressionProtection policy for SQL/XSS."""
    return """<?xml version="1.0" encoding="UTF-8"?>
<RegularExpressionProtection name="Regex-Protection">
    <DisplayName>SQL Injection and XSS Protection</DisplayName>
    <Source>request</Source>
    <QueryParam name="*">
        <Pattern>[\s]*((delete)|(exec)|(drop\s*table)|(insert)|(shutdown)|(update))</Pattern>
        <Pattern>&lt;\\s*script\\b[^&gt;]*&gt;</Pattern>
    </QueryParam>
    <JSONPayload>
        <JSONPath>$.*</JSONPath>
        <Pattern>[\s]*((delete)|(exec)|(drop\s*table)|(insert)|(shutdown)|(update))</Pattern>
    </JSONPayload>
</RegularExpressionProtection>"""


def analyze_apigee_proxy_bundle(bundle_path):
    """Analyze an Apigee proxy bundle for security policy gaps."""
    findings = []
    bundle = Path(bundle_path)
    policies_dir = bundle / "apiproxy" / "policies"
    if not policies_dir.exists():
        return [{"issue": "no_policies_directory", "severity": "CRITICAL"}]
    policy_files = list(policies_dir.glob("*.xml"))
    policy_names = [p.stem for p in policy_files]
    required_policies = [
        ("JSONThreatProtection", "json_threat_protection", "HIGH"),
        ("SpikeArrest", "spike_arrest", "HIGH"),
        ("OAuthV2", "oauth_authentication", "CRITICAL"),
        ("CORS", "cors_policy", "MEDIUM"),
    ]
    for policy_type, issue_name, severity in required_policies:
        has_policy = any(policy_type.lower() in p.lower() for p in policy_names)
        if not has_policy:
            for pf in policy_files:
                content = pf.read_text(errors="ignore")
                if policy_type in content:
                    has_policy = True
                    break
        if not has_policy:
            findings.append({
                "issue": f"missing_{issue_name}",
                "policy_type": policy_type,
                "severity": severity,
            })
    return findings


def audit_threat_protection_config(policy_path):
    """Audit a threat protection policy for weak configurations."""
    findings = []
    content = Path(policy_path).read_text(errors="ignore")
    depth_match = re.search(r"<ContainerDepth>(\d+)</ContainerDepth>", content)
    if depth_match and int(depth_match.group(1)) > 10:
        findings.append({
            "issue": "excessive_container_depth",
            "value": int(depth_match.group(1)),
            "recommended": 5, "severity": "MEDIUM",
        })
    string_match = re.search(r"<StringValueLength>(\d+)</StringValueLength>", content)
    if string_match and int(string_match.group(1)) > 10000:
        findings.append({
            "issue": "excessive_string_length",
            "value": int(string_match.group(1)),
            "recommended": 500, "severity": "MEDIUM",
        })
    rate_match = re.search(r"<Rate>(\d+)(ps|pm)</Rate>", content)
    if rate_match:
        rate_val = int(rate_match.group(1))
        unit = rate_match.group(2)
        if unit == "ps" and rate_val > 100:
            findings.append({
                "issue": "spike_arrest_too_permissive",
                "rate": f"{rate_val}{unit}", "severity": "HIGH",
            })
    return findings


def main():
    parser = argparse.ArgumentParser(description="Apigee API Threat Protection Agent")
    parser.add_argument("--action", choices=[
        "generate", "audit_bundle", "audit_policy", "full"
    ], default="generate")
    parser.add_argument("--bundle", help="Apigee proxy bundle path")
    parser.add_argument("--policy", help="Policy XML file to audit")
    parser.add_argument("--output", default="apigee_threat_protection_report.json")
    args = parser.parse_args()

    report = {"generated_at": datetime.utcnow().isoformat(), "findings": {}}

    if args.action == "generate":
        report["policies"] = {
            "json_threat_protection": generate_json_threat_policy(),
            "spike_arrest": generate_spike_arrest_policy(),
            "regex_protection": generate_regex_protection_policy(),
        }
        print("[+] Generated 3 Apigee security policies")

    if args.action in ("audit_bundle", "full") and args.bundle:
        f = analyze_apigee_proxy_bundle(args.bundle)
        report["findings"]["bundle_audit"] = f
        print(f"[+] Bundle audit findings: {len(f)}")

    if args.action in ("audit_policy", "full") and args.policy:
        f = audit_threat_protection_config(args.policy)
        report["findings"]["policy_audit"] = f
        print(f"[+] Policy audit findings: {len(f)}")

    with open(args.output, "w") as fout:
        json.dump(report, fout, indent=2, default=str)
    print(f"[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()
Keep exploring
Detecting API Enumeration AttacksDetect and prevent API enumeration attacks including BOLA and IDOR exploitation by monitoring sequential identifier access patterns and authorization failures.Implementing API Abuse Detection with Rate LimitingImplement API abuse detection using token bucket, sliding window, and adaptive rate limiting algorithms to prevent DDoS, brute force, and credential stuffing attacks.Implementing API Gateway Security ControlsImplements security controls at the API gateway layer including authentication enforcement, rate limiting, request validation, IP allowlisting, TLS termination, and threat protection. The engineer configures API gateways (Kong, AWS API Gateway, Azure APIM, Apigee) to act as a centralized security enforcement point that validates, throttles, and monitors all API traffic before it reaches backend services. Activates for requests involving API gateway security, API management security, gateway authentication, or centralized API protection.Implementing API Schema Validation SecurityImplement API schema validation using OpenAPI specifications and JSON Schema to enforce input/output contracts and prevent injection, data exposure, and mass assignment attacks.Implementing API Security Testing with 42CrunchImplement comprehensive API security testing using the 42Crunch platform to perform static audit and dynamic conformance scanning of OpenAPI specifications.Testing OAuth2 Implementation FlawsTests OAuth 2.0 and OpenID Connect implementations for security flaws including authorization code interception, redirect URI manipulation, CSRF in OAuth flows, token leakage, scope escalation, and PKCE bypass. The tester evaluates the authorization server, client application, and token handling for common misconfigurations that enable account takeover or unauthorized access. Activates for requests involving OAuth security testing, OIDC vulnerability assessment, OAuth2 redirect bypass, or authorization code flow testing.