Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

NIST CSF 2.0

Overview

Resource Public Key Infrastructure (RPKI) provides cryptographic validation of BGP route origins to prevent route hijacking and accidental route leaks. RPKI enables network operators to create Route Origin Authorizations (ROAs) that declare which Autonomous Systems (ASes) are authorized to originate specific IP prefixes. BGP routers validate received route announcements against RPKI data through Route Origin Validation (ROV), rejecting routes with invalid origins. This skill covers creating ROAs through Regional Internet Registries (RIRs), deploying RPKI validator software, configuring ROV on Cisco IOS-XE and Juniper Junos routers, and implementing BGP filtering policies based on RPKI validation state.

When to Use

When deploying or configuring implementing bgp security with rpki 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

IP address space allocated from an RIR (ARIN, RIPE, APNIC, AFRINIC, LACNIC)
RIR member portal access for ROA creation
BGP routers (Cisco IOS-XE 16.x+, Juniper Junos 12.2+, or similar)
Linux server for RPKI validator/cache (Routinator, FORT, or OctoRPKI)
Understanding of BGP routing and AS path concepts

Core Concepts

RPKI Architecture

┌──────────────────────────────────────────────┐
│           Regional Internet Registries        │
│    (ARIN, RIPE, APNIC, AFRINIC, LACNIC)      │
│                                               │
│  ┌─────────────────────────────────────────┐  │
│  │  Trust Anchor (Root CA Certificate)      │  │
│  │  ├── CA Certificate (ISP/Organization)   │  │
│  │  │   ├── ROA: AS64512 → 198.51.100.0/24 │  │
│  │  │   └── ROA: AS64512 → 2001:db8::/32   │  │
│  │  └── CA Certificate (Another Org)        │  │
│  │      └── ROA: AS64513 → 203.0.113.0/24  │  │
│  └─────────────────────────────────────────┘  │
└──────────────────────────────────────────────┘
                     │ rsync/RRDP
                     ▼
         ┌──────────────────────┐
         │  RPKI Validator/Cache │  (Routinator, FORT, OctoRPKI)
         │  Validates ROAs       │
         │  Serves VRPs to RTR   │
         └──────────────────────┘
                     │ RTR Protocol (TCP 8323)
                     ▼
         ┌──────────────────────┐
         │  BGP Router           │
         │  Performs ROV          │
         │  Applies policy:      │
         │   Valid → Accept      │
         │   Invalid → Reject    │
         │   NotFound → Accept   │
         └──────────────────────┘

RPKI Validation States

State	Meaning	Recommended Action
Valid	ROA exists, origin AS and prefix match	Accept route (prefer)
Invalid	ROA exists, but origin AS or prefix length mismatch	Reject route
NotFound	No ROA covers this prefix	Accept (but lower preference)

Route Origin Authorization (ROA)

A ROA is a signed object that states:

Prefix: The IP address range (e.g., 198.51.100.0/24)
Origin AS: The AS authorized to originate this prefix (e.g., AS64512)
Max Length: Maximum prefix length that can be announced (e.g., /24)

Workflow

Step 1: Create ROAs at Your RIR

ARIN (North America):

Log into ARIN Online portal
Navigate to Routing Security > Route Origin Authorizations
Create ROA:
- Prefix: 198.51.100.0/24
- Origin AS: AS64512
- Max Length: /24 (set equal to prefix length to prevent sub-prefix hijacking)
Sign and submit

RIPE NCC (Europe):

Log into RIPE NCC LIR Portal
Navigate to Certification (RPKI) > ROAs
Create ROA with prefix, origin AS, and max prefix length

Step 2: Deploy RPKI Validator (Routinator)

# Install Routinator on Ubuntu
sudo apt install -y routinator
 
# Or install via Cargo (Rust)
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
cargo install routinator
 
# Initialize Routinator (accept TALs)
routinator init --accept-arin-rpa
 
# Start Routinator in RTR server mode
routinator server \
  --rtr 0.0.0.0:8323 \
  --http 0.0.0.0:8080 \
  --refresh 600 \
  --retry 60 \
  --expire 7200
 
# Run as systemd service
cat > /etc/systemd/system/routinator.service << 'SYSTEMD'
[Unit]
Description=Routinator RPKI Validator
After=network.target
 
[Service]
Type=simple
User=routinator
ExecStart=/usr/bin/routinator server --rtr 0.0.0.0:8323 --http 0.0.0.0:8080
Restart=always
RestartSec=10
 
[Install]
WantedBy=multi-user.target
SYSTEMD
 
sudo systemctl enable routinator
sudo systemctl start routinator
 
# Verify Routinator is serving data
curl http://localhost:8080/api/v1/status
curl http://localhost:8080/api/v1/validity/AS64512/198.51.100.0/24
 
# View Validated ROA Payloads (VRPs)
routinator vrps --format json | head -50

Step 3: Configure ROV on Cisco IOS-XE

! Configure RPKI cache server connection
router bgp 64512
 bgp rpki server tcp 10.0.5.50 port 8323 refresh 600
 
! Verify RPKI session
show bgp rpki server
show bgp rpki table
 
! Create route-map for RPKI-based filtering
route-map RPKI-FILTER permit 10
 match rpki valid
 set local-preference 200
 
route-map RPKI-FILTER permit 20
 match rpki not-found
 set local-preference 100
 
route-map RPKI-FILTER deny 30
 match rpki invalid
 
! Apply to BGP neighbors
router bgp 64512
 address-family ipv4 unicast
  neighbor 198.51.100.1 route-map RPKI-FILTER in
  neighbor 203.0.113.1 route-map RPKI-FILTER in
 
 address-family ipv6 unicast
  neighbor 2001:db8::1 route-map RPKI-FILTER in
 
! Verify ROV operation
show bgp ipv4 unicast rpki validation
show bgp ipv4 unicast 198.51.100.0/24
show ip bgp rpki table
show ip bgp neighbors 198.51.100.1 rpki state

Step 4: Configure ROV on Juniper Junos

# Configure RPKI cache connection
set routing-options validation group RPKI-VALIDATORS session 10.0.5.50 port 8323
set routing-options validation group RPKI-VALIDATORS session 10.0.5.50 refresh-time 600
set routing-options validation group RPKI-VALIDATORS session 10.0.5.50 hold-time 7200
set routing-options validation group RPKI-VALIDATORS session 10.0.5.50 record-lifetime 7200
 
# Create validation policy
set policy-options policy-statement RPKI-POLICY term valid from validation-database valid
set policy-options policy-statement RPKI-POLICY term valid then validation-state valid
set policy-options policy-statement RPKI-POLICY term valid then local-preference 200
set policy-options policy-statement RPKI-POLICY term valid then accept
 
set policy-options policy-statement RPKI-POLICY term invalid from validation-database invalid
set policy-options policy-statement RPKI-POLICY term invalid then validation-state invalid
set policy-options policy-statement RPKI-POLICY term invalid then reject
 
set policy-options policy-statement RPKI-POLICY term unknown from validation-database unknown
set policy-options policy-statement RPKI-POLICY term unknown then validation-state unknown
set policy-options policy-statement RPKI-POLICY term unknown then local-preference 100
set policy-options policy-statement RPKI-POLICY term unknown then accept
 
# Apply to BGP peers
set protocols bgp group TRANSIT import RPKI-POLICY
set protocols bgp group PEERS import RPKI-POLICY
 
# Verify
show validation session
show validation database
show validation statistics
show route validation-state invalid

Step 5: Monitor RPKI Deployment

#!/usr/bin/env python3
"""Monitor RPKI ROV deployment health and coverage statistics."""
 
import json
import sys
import urllib.request
 
 
class RPKIMonitor:
    def __init__(self, routinator_url: str = "http://localhost:8080"):
        self.routinator_url = routinator_url
 
    def get_status(self) -> dict:
        """Get Routinator server status."""
        url = f"{self.routinator_url}/api/v1/status"
        try:
            with urllib.request.urlopen(url) as resp:
                return json.loads(resp.read())
        except Exception as e:
            print(f"Error connecting to Routinator: {e}")
            return {}
 
    def check_validity(self, asn: int, prefix: str) -> dict:
        """Check RPKI validity of a prefix/origin pair."""
        url = f"{self.routinator_url}/api/v1/validity/AS{asn}/{prefix}"
        try:
            with urllib.request.urlopen(url) as resp:
                return json.loads(resp.read())
        except Exception as e:
            return {"error": str(e)}
 
    def get_vrp_count(self) -> int:
        """Get total number of Validated ROA Payloads."""
        status = self.get_status()
        return status.get("vrpsCount", 0)
 
    def report(self, prefixes_to_check: list):
        """Generate RPKI monitoring report."""
        status = self.get_status()
 
        print(f"\n{'='*60}")
        print("RPKI MONITORING REPORT")
        print(f"{'='*60}")
        print(f"\nRoutinator Status:")
        print(f"  Version: {status.get('version', 'Unknown')}")
        print(f"  VRPs Total: {status.get('vrpsCount', 'N/A')}")
        print(f"  Last Update: {status.get('lastUpdateDone', 'N/A')}")
 
        if prefixes_to_check:
            print(f"\nPrefix Validity Checks:")
            for asn, prefix in prefixes_to_check:
                result = self.check_validity(asn, prefix)
                validity = result.get("validated_route", {}).get(
                    "validity", {}).get("state", "error")
                print(f"  AS{asn} -> {prefix}: {validity.upper()}")
 
 
if __name__ == "__main__":
    monitor = RPKIMonitor()
 
    # Check own prefixes
    own_prefixes = [
        (64512, "198.51.100.0/24"),
    ]
 
    monitor.report(own_prefixes)

Best Practices

Create ROAs for All Prefixes - Sign ROAs for every prefix your organization announces
Max Length = Prefix Length - Set max-length equal to announced prefix length to prevent sub-prefix hijacking
Dual Validator - Run two independent RPKI validators for redundancy
Soft Policy First - Start with logging RPKI-invalid routes before dropping them
Monitor ROA Expiry - Set alerts for ROA certificates approaching expiration
Coordinate with Upstreams - Notify transit providers about your RPKI deployment
Test with Looking Glass - Verify your ROAs are visible using public RPKI validators

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.5 KB

API Reference: BGP RPKI Validation Agent

Dependencies

Library	Version	Purpose
requests	>=2.28	HTTP client for RIPEstat and Cloudflare RPKI APIs

CLI Usage

python scripts/agent.py \
  --asn AS13335 \
  --prefixes 1.1.1.0/24 104.16.0.0/12 \
  --output-dir /reports/ \
  --output rpki_report.json

Functions

`validate_prefix_rpki(prefix) -> dict`

Queries RIPEstat /rpki-validation/data.json for RPKI status (valid/invalid/unknown).

`get_roas_for_asn(asn) -> list`

Queries Cloudflare RPKI /api/v1/roas for Route Origin Authorizations.

`get_prefix_overview(prefix) -> dict`

Queries RIPEstat /prefix-overview/data.json for routing overview.

`check_rpki_adoption(asn) -> dict`

Compares announced prefixes against ROA coverage to calculate adoption percentage.

`validate_multiple_prefixes(prefixes) -> list`

Batch validates prefixes against RPKI.

`generate_report(asn, prefixes) -> dict`

Full report with adoption metrics, per-prefix validation, and recommendations.

APIs Used

API	Endpoint
RIPEstat	`stat.ripe.net/data/rpki-validation/data.json`
RIPEstat	`stat.ripe.net/data/announced-prefixes/data.json`
Cloudflare RPKI	`rpki.cloudflare.com/api/v1/roas`

Output Schema

{
  "asn": "AS13335",
  "adoption": {"announced_prefixes": 500, "roa_covered": 498, "coverage_pct": 99.6},
  "prefix_validation": [{"prefix": "1.1.1.0/24", "status": "valid"}],
  "recommendations": ["Create ROAs for 2 uncovered prefixes"]
}

Scripts 1

agent.py4.4 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
"""BGP RPKI validation agent using RIPEstat and Cloudflare RPKI APIs."""

import argparse
import json
import logging
import os
import sys
from datetime import datetime
from typing import List

try:
    import requests
except ImportError:
    sys.exit("requests required: pip install requests")

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

RIPESTAT_BASE = "https://stat.ripe.net/data"
CLOUDFLARE_RPKI = "https://rpki.cloudflare.com/api/v1"


def validate_prefix_rpki(prefix: str) -> dict:
    """Validate a prefix against RPKI using RIPEstat."""
    resp = requests.get(f"{RIPESTAT_BASE}/rpki-validation/data.json",
                        params={"resource": prefix}, timeout=15)
    if resp.status_code == 200:
        data = resp.json().get("data", {})
        return {
            "prefix": prefix,
            "status": data.get("status", "unknown"),
            "validating_roas": data.get("validating_roas", []),
        }
    return {"prefix": prefix, "status": "error"}


def get_roas_for_asn(asn: str) -> List[dict]:
    """Get Route Origin Authorizations for an ASN from Cloudflare RPKI."""
    resp = requests.get(f"{CLOUDFLARE_RPKI}/roas", params={"asn": asn}, timeout=15)
    if resp.status_code == 200:
        return resp.json().get("roas", [])
    return []


def get_prefix_overview(prefix: str) -> dict:
    """Get prefix routing overview from RIPEstat."""
    resp = requests.get(f"{RIPESTAT_BASE}/prefix-overview/data.json",
                        params={"resource": prefix}, timeout=15)
    if resp.status_code == 200:
        return resp.json().get("data", {})
    return {}


def check_rpki_adoption(asn: str) -> dict:
    """Check RPKI adoption status for an ASN."""
    roas = get_roas_for_asn(asn)
    resp = requests.get(f"{RIPESTAT_BASE}/announced-prefixes/data.json",
                        params={"resource": asn}, timeout=15)
    announced = []
    if resp.status_code == 200:
        announced = resp.json().get("data", {}).get("prefixes", [])
    roa_prefixes = {r.get("prefix") for r in roas}
    announced_prefixes = {p.get("prefix") for p in announced}
    covered = announced_prefixes & roa_prefixes
    coverage_pct = (len(covered) / len(announced_prefixes) * 100) if announced_prefixes else 0
    return {
        "asn": asn,
        "announced_prefixes": len(announced_prefixes),
        "roa_covered": len(covered),
        "uncovered": len(announced_prefixes - roa_prefixes),
        "coverage_pct": round(coverage_pct, 1),
        "roa_count": len(roas),
    }


def validate_multiple_prefixes(prefixes: List[str]) -> List[dict]:
    """Validate multiple prefixes against RPKI."""
    results = []
    for prefix in prefixes:
        result = validate_prefix_rpki(prefix)
        results.append(result)
        logger.info("RPKI %s: %s", prefix, result.get("status", "unknown"))
    return results


def generate_report(asn: str, prefixes: List[str]) -> dict:
    """Generate RPKI validation report for an ASN and its prefixes."""
    report = {"analysis_date": datetime.utcnow().isoformat(), "asn": asn}
    report["adoption"] = check_rpki_adoption(asn)
    report["prefix_validation"] = validate_multiple_prefixes(prefixes)
    invalid = [p for p in report["prefix_validation"] if p.get("status") == "invalid"]
    report["invalid_prefixes"] = invalid
    report["recommendations"] = []
    if report["adoption"]["coverage_pct"] < 100:
        report["recommendations"].append(
            f"Create ROAs for {report['adoption']['uncovered']} uncovered prefixes")
    if invalid:
        report["recommendations"].append(f"Investigate {len(invalid)} RPKI-invalid prefixes")
    return report


def main():
    parser = argparse.ArgumentParser(description="BGP RPKI Validation Agent")
    parser.add_argument("--asn", required=True, help="AS number (e.g., AS13335)")
    parser.add_argument("--prefixes", nargs="*", default=[], help="Prefixes to validate")
    parser.add_argument("--output-dir", default=".")
    parser.add_argument("--output", default="rpki_report.json")
    args = parser.parse_args()

    os.makedirs(args.output_dir, exist_ok=True)
    report = generate_report(args.asn, args.prefixes)
    out_path = os.path.join(args.output_dir, args.output)
    with open(out_path, "w") as f:
        json.dump(report, f, indent=2)
    logger.info("Report saved to %s", out_path)
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()

Keep exploring