diff --git a/README.md b/README.md index c2b81b4a..6c8195e6 100644 --- a/README.md +++ b/README.md @@ -88,7 +88,8 @@ _Pass-through management_ interfaces allows the use of the assigned management I NOSes defaulting to _pass-through_ management interfaces are: -* None so far, we are gathering feedback on this, and will update this list as feedback is received. Please contact us in [Discord](https://discord.gg/vAyddtaEV9) or open up an issue here if you have found any issues when trying the passthrough mode. +we are gathering feedback on this, and will update this list as feedback is received. Please contact us in [Discord](https://discord.gg/vAyddtaEV9) or open up an issue here if you have found any issues when trying the passthrough mode. +* Fortinet/Fortigate In case of _host-forwarded_ management interfaces, certain ports are forwarded to the NOS VM IP, which is always 10.0.0.15/24. The management gateway in this case is 10.0.0.2/24, and outgoing traffic is NATed to the container management IP. This management interface connection mode does not allow for traffic such as LLDP to pass through the management interface. diff --git a/fortinet/fortigate/README.md b/fortinet/fortigate/README.md index 398a34c9..3d37dc58 100644 --- a/fortinet/fortigate/README.md +++ b/fortinet/fortigate/README.md @@ -1,20 +1,223 @@ -# Fortinet Fortigate +# Fortinet FortiGate -Support for the Fortinet Fortigate launched by containerlab. +Fortinet FortiGate/FortiOS support for vrnetlab and Containerlab. -## Building the docker image +The launcher supports recent FortiGate VM images, including FortiOS 8.0, and +similar Fortinet CLI families such as FortiProxy where the boot and prompt +patterns are compatible. -Add your qcow2 image to the root of this folder. -Naming format: fortios-vX.Y.Z.qcow2 +## Build -`make` +Place one FortiOS `qcow2` image in this directory. The Makefile expects the +image name to use this format: -## Running the docker image manually +```text +fortios-vX.Y.Z.qcow2 +``` -If you need to run the image without using containerlab: +Build the image: -`make docker-run-fortigate` +```bash +make docker-build-fortigate +``` -## Tested versions +Run the image manually: -* Fortigate 7.0.14 KVM +```bash +make docker-run-fortigate +``` + +## Containerlab + +Use `kind: fortinet_fortigate`. + +```yaml +name: fgt-lab + +topology: + nodes: + fgt: + kind: fortinet_fortigate + image: vr-fortios:8.0.0 + enforce-startup-config: true + startup-config: configs/fgt.conf + license: licenses/appliance.lic + credentials: + username: admin + password: admin + env: + CLAB_MGMT_PASSTHROUGH: "true" +``` + +### Node Options + +`startup-config` mounts a FortiOS config file that the launcher applies after +bootstrap, hostname setup, license handling, admin setup, and baseline config +capture. The file is available inside the container as +`/config/startup-config.cfg`. + +`license` mounts a FortiGate VM license. The launcher expects it inside the +container as `/tftpboot/appliance.lic`, installs it with +`execute restore vmlicense tftp`, handles the reboot, and waits for license +status to leave `Pending`. + +`credentials` sets the desired final administrator account. If omitted, the +final account is `admin` / `admin`. The bootstrap flow handles first-login +password change prompts and FortiOS versions that initially accept a blank +default password. + +`enforce-startup-config: true` is recommended so Containerlab always mounts and +applies the intended startup config. + +## Environment Variables + +| Variable | Default | Values | Description | +| --- | --- | --- | --- | +| `CLAB_MGMT_PASSTHROUGH` | `true` | `true`, `false` | Selects management wiring. `true` uses tap/tc passthrough so the FortiGate management interface participates directly in the Containerlab management network. `false` uses a host-forwarded bridge inside the vrnetlab container. | +| `FOS_DISK_SPECS` | unset | comma-separated `qemu-img create` sizes, for example `10g` or `10g,10g` | Adds extra virtio disks. One disk becomes the FortiGate log disk. Additional disks are formatted during bootstrap; the second disk is expected to become WAN optimization storage on FortiOS versions that support it. | +| `FORTIGATE_UUID` | random UUID | UUID string | Sets the QEMU VM UUID. If unset, a new UUID is generated for each launch. | +| `FOS_LOG_ENCODED` | `false` | `true`, `false` | Logs encoded serial bytes instead of decoded text when enabled. Useful for debugging prompt or terminal parsing issues. | + +Containerlab also passes the usual vrnetlab launch arguments such as hostname, +username, password, and connection mode. For manual runs these are available as +launcher arguments: + +```text +--hostname +--username +--password +--connection-mode +--trace +``` + +## Management Modes + +### Passthrough Management + +`CLAB_MGMT_PASSTHROUGH=true` is the default. The launcher creates a tap device +for `port1` and uses tc rules to redirect management traffic between the +FortiGate VM and the container management interface. TCP serial ports +`5000-5007` are passed through to the container instead of being redirected to +the VM management interface. + +The TFTP server used for license installation runs in a dedicated namespace and +is reachable from the FortiGate through the management gateway address. + +### Host-Forwarded Management + +`CLAB_MGMT_PASSTHROUGH=false` creates an internal `br-mgmt` bridge and configures +FortiGate `port1` with: + +```text +172.31.255.30/30 via 172.31.255.29 +200::1/127 via 200:: +``` + +TCP traffic that enters the container, except the serial console on port `5000`, +is DNATed to the FortiGate management address. UDP traffic is also DNATed so +license TFTP can work. + +If FortiOS later receives DHCP on the management interface, disable the +FortiGate default gateway only after adding a route back to the management +subnet. Disabling it first can cut off management access. + +## Startup Config + +The startup config is applied line by line after the launcher has finished its +own bootstrap commands. Keep it as ordinary FortiOS CLI config: + +```text +config system global + set alias "lab-fgt" +end +``` + +The importer validates basic `config` / `edit` / `next` / `end` nesting and +fails startup on malformed structure. + +The launcher also sets baseline system configuration needed for lab operation, +including management interface addressing, FortiGuard interface selection, DNS, +hostname, and the final administrator account. + +## Licensing + +When `/tftpboot/appliance.lic` exists, the launcher installs it during startup. +License installation may reboot the VM and may remove the active admin session +when the status changes to `VALID`; the launcher handles re-login and continues +bootstrap. + +After installation, the launcher polls `get system status` until the license +field is no longer `Pending` or until the internal +`FOS_LICENSE_STATUS_TIMEOUT_SECONDS` constant expires. + +## Extra Disks + +Set `FOS_DISK_SPECS` to add disks: + +```yaml +env: + FOS_DISK_SPECS: "10g,10g" +``` + +This creates `empty1.qcow2`, `empty2.qcow2`, and so on, and attaches them as +virtio drives. FortiOS normally formats the first additional disk as log +storage. The launcher formats remaining configured disks during bootstrap. + +Expected FortiOS storage usage for common test cases: + +```text +FOS_DISK_SPECS unset -> no configured storage usage +FOS_DISK_SPECS="10g" -> order 1 usage log +FOS_DISK_SPECS="10g,10g" -> order 1 usage log, order 2 usage wanopt +``` + +## Saving Config + +Touch `/get-config` inside a running container to ask the launcher to capture the +current FortiOS config: + +```bash +docker exec clab-- touch /get-config +``` + +The launcher reconnects to the serial console, runs `show`, compares the result +with the baseline captured before startup config application, and writes the +changed config to: + +```text +/config/current.conf +``` + +The serial connection is closed after capture so the console remains available +for external use. If console pagination was enabled before capture, it is +temporarily disabled and then restored. + +## Boot Features + +The FortiOS launcher uses a CLI finite-state machine rather than fixed sleeps. +User-visible behavior includes: + +- detection of login, password, forced password-change, rejected credentials, + welcome banner, reboot, shutdown, and command prompts +- buffered prompt matching for fragmented serial output +- hostname update and prompt-pattern update during bootstrap +- default credential handling across FortiOS 6.4, 7.x, and 8.x behavior +- password-policy failure surfaced as a startup error +- explicit failure if no `qcow2` image is present +- full serial output logging at debug level +- trace logging with `--trace` + +## Tested Versions + +Commit `be1df131b2c000d1ffb79eb941ab8ce4eee07e31` introduced the FortiOS 8.0 +support and was tested with: + +- FortiGate 8.0.0 build 0167 GA debug image +- FortiGate 7.6.6 build 3652 GA debug image +- FortiGate 7.4.12 build 2902 GA +- FortiGate 7.0.19 build 0696 GA +- FortiGate 6.4.16 build 2098 GA +- FortiProxy 7.6.6 build 1628 GA +- FortiProxy 7.4.13 build 0722 GA debug image +- FortiProxy 7.2.16 build 0465 GA +- FortiProxy 7.0.23 build 0222 GA diff --git a/fortinet/fortigate/docker/README.md b/fortinet/fortigate/docker/README.md index a2b03f62..4d75d6e2 100644 --- a/fortinet/fortigate/docker/README.md +++ b/fortinet/fortigate/docker/README.md @@ -1,12 +1,7 @@ -# vrnetlab / Fortinet Fortigaste v7 -======================= -Experimental support for Fortinet fortigate launched by containerlab. +# FortiGate Docker Launcher -## Building the docker image -Add your qcow2 image to the root of this folder. -Naming format: fortios-vX.Y.Z.qcow2 +The public FortiGate/Containerlab API is documented in +[`../README.md`](../README.md). -`make docker-build-fortigate` - -## Running the docker image -`make docker-run-fortigate` +This directory contains the launcher implementation, Dockerfile, and support +modules copied into the vrnetlab image during `make docker-build-fortigate`. diff --git a/fortinet/fortigate/docker/common.py b/fortinet/fortigate/docker/common.py new file mode 100644 index 00000000..3a640419 --- /dev/null +++ b/fortinet/fortigate/docker/common.py @@ -0,0 +1,77 @@ +from enum import IntEnum, auto + + +class FOSCliState(IntEnum): + PROVIDE_USERNAME = 0 + PROVIDE_PASSWORD = auto() + CHANGE_PASSWORD = auto() + CREDENTIAL_REJECTED = auto() + CREDENTIAL_ACCEPTED = auto() + LIC_FAIL = auto() + CMD_PROMPT = auto() + SHUTTING_DOWN = auto() + REBOOTING = auto() + UNKNOWN = auto() # Non-patterns from here on. + TN_TIMEOUT = auto() + + +DEFAULT_HOSTNAME_REGEX = rb"[A-Za-z0-9_.-]+(?:-VM64)?-KVM(?:-[A-Za-z0-9]*)?" +OLD_LIC_HOSTNAME_REGEX = rb"[A-Z]{4,}[0-9]{4,}" +DEFAULT_HOSTNAME_PROMPT = rb"(?m)^\s*" + DEFAULT_HOSTNAME_REGEX + rb"(?:\s+\((?:STS|Interim)\))?\s*[#$]\s*" + +FOS_CLI_STATE_PATTERNS = [None] * FOSCliState.UNKNOWN.value +FOS_CLI_STATE_PATTERNS[FOSCliState.PROVIDE_USERNAME.value] = ( + rb"\n" + DEFAULT_HOSTNAME_REGEX + + rb"(?:\((?:Primary|Secondary)\))?" + rb"\s+login:\s*" +) +FOS_CLI_STATE_PATTERNS[FOSCliState.CHANGE_PASSWORD.value] = b"(?m)^New Password:" +FOS_CLI_STATE_PATTERNS[FOSCliState.PROVIDE_PASSWORD.value] = b"(?m)^Password:" +FOS_CLI_STATE_PATTERNS[FOSCliState.CREDENTIAL_REJECTED.value] = rb"(?m)^Login incorrect\r?$" +FOS_CLI_STATE_PATTERNS[FOSCliState.CREDENTIAL_ACCEPTED.value] = rb"(?m)^Welcome ?!\r?$" +FOS_CLI_STATE_PATTERNS[FOSCliState.LIC_FAIL.value] = rb"(?m)^VM license install failed.\r$" +FOS_CLI_STATE_PATTERNS[FOSCliState.CMD_PROMPT.value] = DEFAULT_HOSTNAME_PROMPT +FOS_CLI_STATE_PATTERNS[FOSCliState.SHUTTING_DOWN.value] = b"system is going down" +FOS_CLI_STATE_PATTERNS[FOSCliState.REBOOTING.value] = b"stand by while rebooting" + +DEF_POLICY_COMPLIANT_PASSWORD = "FortinetFOS1!" +DEFAULT_USERNAME = "admin" +DEFAULT_PASSWORD = DEFAULT_USERNAME + + +class Credentials: + def __init__(self, username, password) -> None: + super().__init__() + self.username = username + self.password = password + + +class LineBuffer: + def __init__(self, lines=2, max_buffer=1024) -> None: + super().__init__() + if lines < 1: + raise ValueError("lines must be at least 1") + self._lines = lines + self._max_buffer = max_buffer + self._data = b"" + + @property + def data(self): + return self._data + + def put(self, data): + if not data: + return + self._data += data + newline_indexes = [idx for idx, byte in enumerate(self._data) if byte == ord("\n")] + keep_from_index = -1 + if self._data.endswith(b"\n") and len(newline_indexes) > self._lines: + keep_from_index = newline_indexes[-(self._lines + 1)] + elif not self._data.endswith(b"\n") and len(newline_indexes) >= self._lines: + keep_from_index = newline_indexes[-self._lines] + if keep_from_index >= 0: + self._data = self._data[keep_from_index:] + self._data = self._data[-self._max_buffer:] + + def clear(self): + self._data = b"" diff --git a/fortinet/fortigate/docker/fos_cli_driver.py b/fortinet/fortigate/docker/fos_cli_driver.py new file mode 100644 index 00000000..4470ef5e --- /dev/null +++ b/fortinet/fortigate/docker/fos_cli_driver.py @@ -0,0 +1,192 @@ +import os +import re +import time + +import vrnetlab +from common import FOSCliState, FOS_CLI_STATE_PATTERNS, Credentials, DEFAULT_USERNAME, DEF_POLICY_COMPLIANT_PASSWORD, \ + DEFAULT_PASSWORD, LineBuffer +from fos_commander import FOSCommander + + +class FOSCliDriver: + """ + Drives the CLI through boot, login, and ready-for-cmd states. + """ + + def __init__(self, terminal: vrnetlab.VM, mgmt_passthrough, username, password, logger, mgmt_address_ipv4, + mgmt_gw_ipv4, mgmt_address_ipv6, mgmt_gw_ipv6, hostname) -> None: + super().__init__() + self._idle_spins = 0 + self._logger = logger + pwd = password + if pwd is None: # emtpy string is falsy, so the or trick doesn't work. + pwd = DEFAULT_PASSWORD + self._desired_credentials = Credentials(username or DEFAULT_USERNAME, pwd) + self._credentials = Credentials(DEFAULT_USERNAME, DEF_POLICY_COMPLIANT_PASSWORD) + self._username = DEFAULT_USERNAME + self._terminal = terminal + self._mgmt_passthrough = mgmt_passthrough + self._log_encoded = os.getenv("FOS_LOG_ENCODED", "false").lower() == "true" + self._state_handlers = { + FOSCliState.PROVIDE_USERNAME: self._provide_username, + FOSCliState.PROVIDE_PASSWORD: self._provide_password, + FOSCliState.CHANGE_PASSWORD: self._change_password, + FOSCliState.CREDENTIAL_REJECTED: self._credential_rejected, + FOSCliState.CREDENTIAL_ACCEPTED: self._credential_accepted, + FOSCliState.LIC_FAIL: self._license_fail, + FOSCliState.CMD_PROMPT: self._cmd_prompt, + FOSCliState.SHUTTING_DOWN: self._shutting_down, + FOSCliState.REBOOTING: self._rebooting, + FOSCliState.UNKNOWN: self._unknown_state, + FOSCliState.TN_TIMEOUT: self._tn_timeout, + } + + self._state_patterns = FOS_CLI_STATE_PATTERNS.copy() + self._line_buffer = LineBuffer() + self.tn_out = b"" + self._last_known_state = FOSCliState.UNKNOWN + self._waiting_for = [] + self._commander = FOSCommander( + terminal=terminal, + logger=logger, + mgmt_address_ipv4=mgmt_address_ipv4, + mgmt_gw_ipv4=mgmt_gw_ipv4, + mgmt_address_ipv6=mgmt_address_ipv6, + mgmt_gw_ipv6=mgmt_gw_ipv6, + mgmt_passthrough=mgmt_passthrough, + hostname=hostname, + state_patterns=self._state_patterns, + waiting_for_state=self._waiting_for, + credentials=self._credentials, + desired_credentials=self._desired_credentials, + ) + + self._tried_v7_default_password = False + self._cred_rejected = False + + def process_state(self): + spin_start = time.time() + # Running signals health state. Stopped signals we stopped before reaching healthy state + while not self._terminal.stopped and time.time() < spin_start + 5: + if self._idle_spins > 300: + # too many spins without appropriate communication + self._logger.warning("no output from serial console, restarting VCP") + self._terminal.stop() + self._idle_spins = 0 + raise RuntimeError("VM node malfunction") + cur_state = self._next_state() + + # The FSM is actually moving along + if cur_state.value < FOSCliState.UNKNOWN.value: + self._idle_spins = 0 + self._last_known_state = cur_state + + # Continue to show current state while reducing verbosity + if cur_state != FOSCliState.UNKNOWN and cur_state != FOSCliState.TN_TIMEOUT: + self._logger.debug(f"ST: {cur_state.name}") + + if len(self.tn_out) > 0: + log_out = self.tn_out + if not self._log_encoded: + log_out = log_out.decode() + self._logger.debug(f"OUT: {log_out}") + + if self._waiting_for and cur_state not in self._waiting_for: + return + + self._waiting_for.clear() + self._state_handlers[cur_state]() + if self._terminal.running: + return # If reached running state, then we allow vrnetlab.VM to be more responsive to system state + + def _next_state(self): + (_, _, res) = self._terminal.tn.expect(self._state_patterns, 1) + self._line_buffer.put(res) + + ridx, match = self._match_buffered_state() + self.tn_out = res + if match: + self.tn_out = self._line_buffer.data[:match.end()] + self._line_buffer.clear() + return FOSCliState(ridx) + if res == b"": + return FOSCliState.TN_TIMEOUT + return FOSCliState.UNKNOWN + + def _match_buffered_state(self): + for ridx, pattern in enumerate(self._state_patterns): + match = re.search(pattern, self._line_buffer.data) + if match: + return ridx, match + return -1, None + + def _provide_username(self): + self._terminal.wait_write(self._credentials.username, wait=None) + + def _provide_password(self): + if self._cred_rejected: + self._tried_v7_default_password = True + self._terminal.wait_write("", wait=None) + return + self._terminal.wait_write(self._credentials.password, wait=None) + + def _change_password(self): + # At this time, this would be the default password. + self._terminal.wait_write(self._credentials.password, wait=None) + self._terminal.wait_write(self._credentials.password, wait="Confirm Password") + # FOS 7.4 needs log out before you can use the password with ssh + self._commander.logout() + + def save_config(self): + self._commander.save_config() + self._reactivate() + + def _reactivate(self): + self._last_known_state = FOSCliState.UNKNOWN + self._line_buffer.clear() + + def _credential_accepted(self): + self._cred_rejected = False + self._tried_v7_default_password = False + + def _cmd_prompt(self): + self._commander.run_cmd() + + def _shutting_down(self): + pass + + def _rebooting(self): + pass + + def _credential_rejected(self): + if not self._tried_v7_default_password: + self._logger.debug("Credential rejected. Possibly never configured. Trying default password next time.") + self._cred_rejected = True + return + additional_info = "" + if b"pasword policy" in self.tn_out: + additional_info = "Min password policy not met. Check the logs for the password policy" + self._logger.error(f"Credential rejected. {additional_info}") + + self._terminal.stop() + raise RuntimeError("Credential rejected") + + def _license_fail(self): + self._logger.error("Failed to setup license.") + self._terminal.stop() + raise RuntimeError("License setup failed") + + def _unknown_state(self): + # no match, if we saw some output from the router it's probably + # booting, so let's give it some more time + if self._last_known_state == FOSCliState.REBOOTING: + self._idle_spins += 1 + # It could be that the FGT image was defective. In this case it has been observed that + # the FGT would endlessly reboot. + else: + self._idle_spins = 0 + + def _tn_timeout(self): + if self._last_known_state == FOSCliState.CMD_PROMPT: + self._cmd_prompt() + self._idle_spins += 1 diff --git a/fortinet/fortigate/docker/fos_commander.py b/fortinet/fortigate/docker/fos_commander.py new file mode 100644 index 00000000..5ac18923 --- /dev/null +++ b/fortinet/fortigate/docker/fos_commander.py @@ -0,0 +1,715 @@ +import datetime +import difflib +import os +import re +import select +import time +from collections import deque + +import vrnetlab +from common import FOSCliState, OLD_LIC_HOSTNAME_REGEX, DEFAULT_HOSTNAME_REGEX, Credentials, LineBuffer + +STARTUP_CONFIG_FILE = "/config/startup-config.cfg" +INIT_CONFIG_FILE = "/tmp/initial.conf" +CURRENT_CONFIG_FILE = "/config/current.conf" +CURRENT_RAW_FILE = "/tmp/current.raw" +CURRENT_CLEAN_FILE = "/tmp/current.clean" +LIC_FILE = "appliance.lic" +CONFIG_CAPTURE_TIMEOUT = 60 +CURRENT_PASSWORD_PATTERN = rb"(?mi)^(?:Please enter current administrator password|Current Password):?\s*$" +LICENSE_STATUS_PATTERN = rb"(?mi)^License(?: Status)?:\s*(.+?)\s*\r?$" +LICENSE_STATUS_PENDING = "pending" +FOS_LICENSE_STATUS_TIMEOUT_SECONDS = 90 +LICENSE_STATUS_POLL_INTERVAL_SECONDS = 2 +ADMIN_SESSIONS_REMOVED_PATTERN = ( + rb"(?m)^\*ATTENTION\*: Admin sessions removed because license registration status changed.*\r?$" +) + + +class TelnetCaptureSession: + def __init__(self, tn): + self._tn = tn + + def read_available(self, timeout): + data = self._tn.read_very_eager() + if data: + return data + + readable, _, _ = select.select([self._tn.get_socket()], [], [], timeout) + if not readable: + return b"" + return self._tn.read_very_eager() + + def write(self, data): + self._tn.write(data) + + def close(self): + pass + + +class FOSCommander: + """ + Dispatches commands after login to FOS CLI. + """ + + def __init__(self, terminal: vrnetlab.VM, logger, mgmt_address_ipv4, mgmt_gw_ipv4, mgmt_address_ipv6, mgmt_gw_ipv6, + mgmt_passthrough, hostname, waiting_for_state, state_patterns, credentials: Credentials, + desired_credentials: Credentials) -> None: + super().__init__() + self._desired_credentials = desired_credentials + self._credentials = credentials + self._state_patterns = state_patterns + self._waiting_for_state: list = waiting_for_state + self._hostname = hostname + self._mgmt_passthrough = mgmt_passthrough + self._mgmt_gw_ipv6 = mgmt_gw_ipv6 + self._mgmt_address_ipv6 = mgmt_address_ipv6 + self._mgmt_gw_ipv4 = mgmt_gw_ipv4 + self._mgmt_address_ipv4 = mgmt_address_ipv4 + self._logger = logger + self._terminal = terminal + self._admin_password_buffer = LineBuffer() + self._start_time = datetime.datetime.now() + self._cmd_queue = deque() + self._disks_formatted = 0 + # The first additional disk is always automatically formatted. + self._disks_to_format = max(0, len(os.getenv("FOS_DISK_SPECS", "").split(",")) - 1) + self._cmd_queue.append(lambda: self._toggle_paging(False)) + if self._disks_to_format > self._disks_formatted: + self._cmd_queue.appendleft(self._format_next_disk) + self._cmd_queue.append(self._configure_sys_if) + self._cmd_queue.append(self._setup_default_dns) + self.check_license_exists() + self._cmd_queue.append(self._update_hostname) + self._cmd_queue.append(self._add_admin) + self._cmd_queue.append(self._capture_blank_config) + self._cmd_queue.append(self._apply_startup_config) + + def run_cmd(self): + try: + next_cmd = self._cmd_queue.popleft() + except IndexError: + self._cmd_queue.append(self._ready) + return + next_cmd() + + def logout(self): + self._cmd_queue.appendleft(lambda: self._terminal.wait_write("exit", wait=None)) + + def save_config(self): + restore_paging = False + try: + self._wait_for_prompt_sync("Timed out waiting for command prompt before config capture.") + restore_paging = self._console_paging_enabled() + if restore_paging: + self._toggle_paging(False) + self._save_config() + finally: + try: + if restore_paging: + self._toggle_paging(True) + finally: + self._terminal.tn.close() + + def _format_next_disk(self): + if self._disks_to_format == self._disks_formatted: + self._logger.info("Done formatting disks.") + return + + disk_number = self._disks_formatted + 2 + self._logger.info(f"Formatting disk #{disk_number}") + self._terminal.wait_write("exe disk list", wait=None) + disk_list_output, complete = self._read_until_pattern( + self._state_patterns[FOSCliState.CMD_PROMPT.value], + time.monotonic() + 10, + ) + # We took the cmd prompt from the buffer, this regenerates it so the FOSCliDriver can detect it. + self._terminal.wait_write("", wait=None) + if not complete: + self._logger.error("Timed out waiting for disk list output.") + self._terminal.stop() + raise RuntimeError("Timed out waiting for disk list output.") + + disk_ref = self._disk_ref_from_list(disk_list_output, disk_number) + self._terminal.wait_write(f"exe disk format {disk_ref}", wait=None) + self._terminal.wait_write(f"y", wait="continue") + self._disks_formatted += 1 + if self._disks_to_format > self._disks_formatted: + self._cmd_queue.appendleft(self._format_next_disk) + self._waiting_for_state.clear() + self._waiting_for_state.extend([FOSCliState.REBOOTING]) + + def _disk_ref_from_list(self, disk_list_output, disk_number): + disk_name = f"Virtual-Disk{disk_number}".encode() + disk_line_match = re.search( + rb"(?m)^Disk\s+" + re.escape(disk_name) + rb"\s+ref:\s+(\d+)\b.*$", + disk_list_output, + ) + if not disk_line_match: + self._logger.error( + f"Could not find {disk_name.decode()} in disk list output: {disk_list_output.decode(errors='replace')}" + ) + self._terminal.stop() + raise RuntimeError(f"Could not find {disk_name.decode()} in disk list output.") + return disk_line_match.group(1).decode() + + def _setup_default_dns(self): + self._terminal.wait_write("config system dns\r" + "set primary 1.1.1.1\r" + "set secondary 8.8.8.8\r" + "end", + wait=None) + + def _toggle_paging(self, enabled): + output_mode = "more" if enabled else "standard" + command = "config system console\r" f"set output {output_mode}\r" "end" + self._terminal.tn.write((command + "\r").encode()) + + def _console_paging_enabled(self): + self._terminal.tn.write(b"show full-configuration system console\r") + output, complete = self._read_until_pattern( + self._state_patterns[FOSCliState.CMD_PROMPT.value], + time.monotonic() + 10, + ) + if not complete: + self._terminal.stop() + raise RuntimeError("Timed out reading console output mode.") + match = re.search(rb"(?m)^\s*set output (more|standard)\s*\r?$", output) + if not match: + self._logger.warn("Could not determine console output mode; not restoring pagination after capture.") + return False + return match.group(1) == b"more" + + def _wait_for_prompt_sync(self, timeout_message): + _output, complete = self._read_until_pattern( + self._state_patterns[FOSCliState.CMD_PROMPT.value], + time.monotonic() + 10, + ) + if not complete: + self._terminal.stop() + raise RuntimeError(timeout_message) + + def _unset_default_dns(self): + self._terminal.wait_write("config system dns\r" + "set primary 1.1.1.1\r" + "set secondary 8.8.8.8\r" + "end", + wait=None) + + def _configure_sys_if(self): + if self._mgmt_address_ipv4 == "dhcp": + self._logger.info("MGMT IP is in DHCP mode") + return + self._logger.info(f"Setting mgmt IPv4={self._mgmt_address_ipv4} and IPv6={self._mgmt_address_ipv6}") + self._terminal.wait_write("config system interface\r" + "edit port1\r" + "set mode static\r" + f"set ip {self._mgmt_address_ipv4}\r" + "set allowaccess ping https ssh http", + wait=None) + if self._mgmt_address_ipv6 is not None: + self._terminal.wait_write("config ipv6\r" + "set ip6-mode static\r" + f"set ip6-address {self._mgmt_address_ipv6}\r" + "set ip6-allowaccess ping https ssh http\r" + "end", + wait="allowaccess") + self._terminal.wait_write("next\r" + "end", + wait="http") + + self._terminal.wait_write("config system fortiguard\r" + "set interface-select-method specify\r" + "set interface port1\r" + "end", wait="end") + self._terminal.wait_write("config router static\r" + "ed 9999\r" + f"set gateway {self._mgmt_gw_ipv4}\r" + "set device port1\r" + "next\r" + "end", + wait="end") + if self._mgmt_address_ipv6 is not None: + self._terminal.wait_write("config router static6\r" + "ed 9999\r" + f"set gateway {self._mgmt_gw_ipv6}\r" + "set device port1\r" + "next\r" + "end", + wait=None) + self._terminal.wait_write("", wait="end") + + def _update_now(self): + self._terminal.wait_write("exe update-now", wait=None) + echo_result = self._wait_for_command_echo("exe update-now", time.monotonic() + 5) + if echo_result != "echo": + self._logger.warn("Timed out waiting for update-now command echo.") + return + res = self._enter_current_password_if_asked() + if res == "session_lost": + self._logger.warn("Timed out waiting for update-now to return to the command prompt.") + return + + if not self._wait_for_license_status_ready(): + self._logger.warn("Timed out waiting for license status to change.") + return + self._logger.info("License status changed") + + def _wait_for_license_status_ready(self): + deadline = time.monotonic() + FOS_LICENSE_STATUS_TIMEOUT_SECONDS + last_status = None + + while time.monotonic() < deadline: + self._terminal.wait_write("get system status", wait=None) + echo_result = self._wait_for_command_echo( + "get system status", + deadline, + session_loss_is_terminal=True, + ) + if echo_result == "session_lost": + return True + if echo_result != "echo": + self._logger.warn("Timed out waiting for system status command echo.") + return False + status_output, ridx, complete = self._read_until_patterns( + [self._state_patterns[FOSCliState.CMD_PROMPT.value]], + deadline, + extra_patterns=self._license_session_loss_patterns(), + ) + if not complete: + self._logger.warn("Timed out waiting for system status output.") + return False + + session_removed = re.search(ADMIN_SESSIONS_REMOVED_PATTERN, status_output) + status = self._license_status_from_system_status(status_output) + if status: + last_status = status + self._logger.info(f"License status is {status}") + if session_removed: + self._handle_license_session_removed() + return True + if status.lower() != LICENSE_STATUS_PENDING: + self._terminal.wait_write("", wait=None) + return True + else: + self._logger.warn("Could not find License field in system status output.") + + if ridx > 0 or session_removed: + self._handle_license_session_removed() + return True + + remaining = deadline - time.monotonic() + if remaining <= 0: + break + time.sleep(min(LICENSE_STATUS_POLL_INTERVAL_SECONDS, remaining)) + + if last_status: + self._logger.warn(f"License status remained {last_status}.") + self._terminal.wait_write("", wait=None) + return False + + def _read_until_pattern(self, pattern, deadline): + output, _, complete = self._read_until_patterns([pattern], deadline) + return output, complete + + def _read_until_patterns(self, end_patterns, deadline, extra_patterns=None): + output = b"" + extra_patterns = extra_patterns or [] + patterns = list(end_patterns) + list(extra_patterns) + end_pattern_count = len(end_patterns) + + while time.monotonic() < deadline: + remaining = deadline - time.monotonic() + (ridx, match, chunk) = self._expect(patterns, min(10, remaining)) + output += chunk + if not match: + return output, -1, False + if ridx < end_pattern_count: + return output, ridx, True + return output, ridx, True + + return output, -1, False + + def _wait_for_command_echo(self, command, deadline, session_loss_is_terminal=False): + command_pattern = re.escape(command.encode()) + session_loss_patterns = self._license_session_loss_patterns() if session_loss_is_terminal else [] + patterns = session_loss_patterns + [command_pattern] + while time.monotonic() < deadline: + remaining = deadline - time.monotonic() + (ridx, match, _) = self._expect(patterns, min(5, remaining)) + if not match: + continue + if ridx < len(session_loss_patterns): + self._handle_license_session_removed() + return "session_lost" + return "echo" + return "timeout" + + def _license_session_loss_patterns(self): + return [ + ADMIN_SESSIONS_REMOVED_PATTERN, + self._state_patterns[FOSCliState.PROVIDE_USERNAME.value], + self._state_patterns[FOSCliState.PROVIDE_PASSWORD.value], + ] + + def _handle_license_session_removed(self): + self._logger.info("License registration removed the admin session; waiting for login.") + self._clear_stale_license_poll_login() + self._waiting_for_state.clear() + self._waiting_for_state.extend([FOSCliState.PROVIDE_USERNAME]) + + def _clear_stale_license_poll_login(self): + # The poll command can race with FortiOS removing the admin session. In + # that case "get system status" is accepted as a username, and FortiOS + # waits at Password:. Clear that failed login before returning to the FSM. + deadline = time.monotonic() + 5 + saw_password = False + patterns = [ + self._state_patterns[FOSCliState.PROVIDE_PASSWORD.value], + self._state_patterns[FOSCliState.PROVIDE_USERNAME.value], + ] + + while time.monotonic() < deadline: + remaining = deadline - time.monotonic() + ridx, match, _ = self._expect(patterns, min(1, remaining)) + if not match: + continue + if ridx == 0: + self._terminal.wait_write("", wait=None) + saw_password = True + break + + if not saw_password: + self._terminal.wait_write("", wait=None) + + def _license_status_from_system_status(self, status_output): + match = re.search(LICENSE_STATUS_PATTERN, status_output) + if not match: + return None + return match.group(1).decode(errors="replace").strip() + + def _expect(self, patterns, timeout): + result = self._terminal.tn.expect(patterns, timeout) + output = result[2] + self._logger.debug(f"OUT: {output.decode(errors='replace')}") + return result + + def _move_mgmt_to_vrf1(self): + self._terminal.wait_write("config system interface\r" + "edit port1", + wait=None) + self._terminal.wait_write("set vrf 1\r" + "next\r" + "end", wait="port1") + self._terminal.wait_write("config router static\r" + "edit 9999\r" + "set vrf 1\r" + "next\r" + "end", wait=None) + if self._mgmt_address_ipv6 is not None: + self._terminal.wait_write("config router static6\r" + "edit 9999\r" + "set vrf 1\r" + "next\r" + "end", wait=None) + + def _update_hostname(self): + self._terminal.wait_write("config system global", wait=None) + hostname_command = "set hostname " + self._hostname + self._terminal.wait_write(hostname_command, wait="global") + self._terminal.wait_write("end", wait=hostname_command) + self._state_patterns[FOSCliState.PROVIDE_USERNAME.value] = ( + rb"\n" + self._hostname.encode("utf-8") + + rb"(?:\((?:Primary|Secondary)\))?" + + rb"\s+login:\s*") + self._state_patterns[FOSCliState.CMD_PROMPT.value] = ( + rb"(?m)^ ?" + + re.escape(self._hostname.encode("utf-8")) + + rb" ?" + + rb"(?:\s+\((?:STS|Interim)\))?" + + rb" ?[#$] ?" + ) + + def _setup_license(self): + if self._mgmt_passthrough: + tftp_server_ip = self._mgmt_gw_ipv4 + else: + tftp_server_ip = self._mgmt_gw_ipv4 + + self._logger.info(f"Setting up license {LIC_FILE} from server {tftp_server_ip}") + + self._terminal.wait_write(f"exe restore vmlicense tftp {LIC_FILE} {tftp_server_ip}", wait=None) + self._terminal.wait_write("y", wait="Do you want to continue?") + self._state_patterns[FOSCliState.PROVIDE_USERNAME.value] = ( + rb"\n(?:" + OLD_LIC_HOSTNAME_REGEX + b"|" + DEFAULT_HOSTNAME_REGEX + rb") ?" + + rb"(?:\((?:Primary|Secondary)\))?" + + rb"\s+login:\s*") + self._state_patterns[FOSCliState.CMD_PROMPT.value] = ( + rb"(?m)^ ?" + + rb"(?:" + OLD_LIC_HOSTNAME_REGEX + b"|" + DEFAULT_HOSTNAME_REGEX + b")" + + rb" ?" + + rb"(?:\s+\((?:STS|Interim)\))?" + + rb" ?[#$] ?" + ) + self._waiting_for_state.clear() + self._waiting_for_state.extend([FOSCliState.REBOOTING, FOSCliState.LIC_FAIL]) + self._cmd_queue.appendleft(self._unset_default_dns) + self._cmd_queue.appendleft(self._move_mgmt_to_vrf1) + self._cmd_queue.appendleft(self._update_now) + self._cmd_queue.appendleft(self._configure_sys_if) # Was seeing static route disappear. + + def _apply_startup_config(self): + """Load additional config provided by user.""" + + if not os.path.exists(STARTUP_CONFIG_FILE): + self._logger.trace(f"Startup config file {STARTUP_CONFIG_FILE} is not found") + return + + self._logger.trace(f"Configuring with startup-config from file: {STARTUP_CONFIG_FILE}") + config_lines = [] + with open(STARTUP_CONFIG_FILE) as file: + config_lines = file.readlines() + + config_stack = deque() + wait_for = None + for line in config_lines: + r_stripped = line.rstrip() + full_stripped = r_stripped.lstrip() + _wait_for = None + if full_stripped.startswith("config"): + config_stack.append("c") + if full_stripped.startswith("edit"): + config_stack.append("e") + if full_stripped.startswith("next") or full_stripped.startswith("end"): + top = config_stack.pop() + _wait_for = full_stripped + if full_stripped.startswith("next") and top != "e": + self._logger.error("Startup config malformed. \"next\" command outside of edit scope.") + raise ValueError("Startup config malformed. \"next\" command outside of edit scope.") + if full_stripped.startswith("end") and top != "c": + self._logger.error("Startup config malformed. \"end\" command outside of config scope.") + raise ValueError("Startup config malformed. \"end\" command outside of config scope.") + # Maintaining nesting produces a more appealing debug log. + self._terminal.wait_write(r_stripped, wait=wait_for) + wait_for = _wait_for + + if len(config_stack) > 0: + raise ValueError("Startup config malformed. Unmatched config or edit brackets.") + + def capture_config(self): + self._logger.info("Capturing FortiOS config with telnet plain show") + session = TelnetCaptureSession(self._terminal.tn) + return self._capture_config_from_session(session) + + def _capture_blank_config(self): + config = self.capture_config() + self._write_config_file(INIT_CONFIG_FILE, config) + + def _save_config(self): + blank_config = self._read_config_file(INIT_CONFIG_FILE) + current_config = self.capture_config() + self._write_config_file(CURRENT_CLEAN_FILE, current_config) + changed_config = self._current_side_config_delta(blank_config, current_config) + self._write_config_file(CURRENT_CONFIG_FILE, changed_config) + self._logger.debug("Current config written") + + def _current_side_config_delta(self, blank_config, current_config): + blank_lines = self._normalize_config_for_diff(blank_config) + current_entries = self._config_diff_entries(current_config) + current_lines = [normalized for normalized, _original in current_entries] + matcher = difflib.SequenceMatcher(a=blank_lines, b=current_lines, autojunk=False) + changed_lines = [] + + for tag, _i1, _i2, j1, j2 in matcher.get_opcodes(): + if tag in ("insert", "replace"): + changed_lines.extend(original for _normalized, original in current_entries[j1:j2]) + + if not self._has_non_structural_config_line(changed_lines): + return "" + return "\n".join(changed_lines) + + def _config_diff_entries(self, config): + entries = [] + for line in config.splitlines(): + normalized = self._normalize_config_line(line) + if normalized and not self._is_volatile_config_line(normalized): + entries.append((normalized, line.rstrip())) + return entries + + def _normalize_config_for_diff(self, config): + lines = [] + for line in config.splitlines(): + normalized = self._normalize_config_line(line) + if normalized and not self._is_volatile_config_line(normalized): + lines.append(normalized) + return lines + + def _normalize_config_line(self, line): + return re.sub(r"[ \t]+", " ", line).strip() + + def _is_volatile_config_line(self, line): + return line.startswith("#conf_file_ver=") or line.startswith("conf_file_ver=") + + def _has_non_structural_config_line(self, lines): + for line in lines: + stripped = line.strip() + if not stripped: + continue + if stripped in ("next", "end"): + continue + if stripped.startswith("config ") or stripped.startswith("edit "): + continue + return True + return False + + def _clean_show_output(self, output): + config = output.replace("\r\n", "\n") + config = config.replace("\r", "\n") + config = config.replace("^H", "") + config = re.sub(r"\x08+", "", config) + config = re.sub(r"\x1b\[[0-9;?]*[ -/]*[@-~]", "", config) + lines = config.splitlines() + if lines and lines[0].strip() == "show": + lines = lines[1:] + return "\n".join(lines).strip() + + def _capture_config_from_session(self, session): + prompt_pattern = self._state_patterns[FOSCliState.CMD_PROMPT.value] + try: + session.write(b"show\r") + output = self._read_show_output(session, prompt_pattern) + self._write_config_file(CURRENT_RAW_FILE, output.decode(errors="replace")) + return self._clean_show_output(output.decode(errors="replace")) + finally: + session.close() + + def _read_show_output(self, session, prompt_pattern): + output = b"" + deadline = time.monotonic() + CONFIG_CAPTURE_TIMEOUT + + while True: + if time.monotonic() > deadline: + self._terminal.stop() + raise RuntimeError("Timed out waiting for config capture output.") + + chunk = session.read_available(1) + if not chunk: + continue + + output += chunk + if re.search(prompt_pattern, output): + return re.sub(prompt_pattern + rb"\s*$", b"", output) + + def _read_config_file(self, path): + with open(path) as config_file: + return config_file.read() + + def _write_config_file(self, path, content): + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as config_file: + config_file.write(content) + if content and not content.endswith("\n"): + config_file.write("\n") + + def check_license_exists(self): + try: + os.stat(f"/tftpboot/{LIC_FILE}") + self._cmd_queue.append(self._setup_license) + except FileNotFoundError: + pass + + def _add_admin(self): + username = self._desired_credentials.username + password = self._desired_credentials.password + self._logger.info(f"Configuring admin '{username}'") + + self._terminal.wait_write("config system password-policy\r" + "set status disable\r" + "end", + wait=None) + + self._terminal.wait_write("config system admin", wait=None) + self._terminal.wait_write(f"edit {username}", wait=None) + self._terminal.wait_write("set accprofile super_admin", wait=None) + if len(password) > 0: + self._terminal.wait_write(f"set password {password}", wait="super_admin") + session_state = self._enter_current_password_if_asked() + if self._resume_after_admin_session_loss(session_state): + return + elif username == "admin": + self._terminal.wait_write(f"unset password", wait="super_admin") + session_state = self._enter_current_password_if_asked() + if self._resume_after_admin_session_loss(session_state): + return + + self._terminal.wait_write("next", wait=None) + session_state = self._enter_current_password_if_asked() + if self._resume_after_admin_session_loss(session_state): + return + + # end returns to the top-level prompt; leave it in the buffer so the FSM + # picks it up as CMD_PROMPT on the next spin. + self._terminal.wait_write("end", wait=None) + + # From now on the active admin is the one we just configured. + self._activate_desired_credentials() + + def _activate_desired_credentials(self): + self._credentials.username = self._desired_credentials.username + self._credentials.password = self._desired_credentials.password + + def _resume_after_admin_session_loss(self, session_state): + if session_state not in ("session_lost", "login_prompt_seen", "password_prompt_seen"): + return False + + self._activate_desired_credentials() + self._waiting_for_state.clear() + if session_state == "login_prompt_seen": + self._terminal.wait_write(self._credentials.username, wait=None) + self._waiting_for_state.extend([FOSCliState.PROVIDE_PASSWORD]) + elif session_state == "password_prompt_seen": + self._terminal.wait_write(self._credentials.password, wait=None) + self._waiting_for_state.extend([FOSCliState.CMD_PROMPT]) + else: + self._waiting_for_state.extend([FOSCliState.PROVIDE_USERNAME]) + return True + + def _enter_current_password_if_asked(self): + patterns = [ + CURRENT_PASSWORD_PATTERN, + ADMIN_SESSIONS_REMOVED_PATTERN, + self._state_patterns[FOSCliState.PROVIDE_USERNAME.value], + self._state_patterns[FOSCliState.PROVIDE_PASSWORD.value], + ] + (ridx, _, res) = self._expect(patterns, 2) + self._admin_password_buffer.put(res) + if ridx in (1, 2, 3) or re.search(ADMIN_SESSIONS_REMOVED_PATTERN, self._admin_password_buffer.data): + self._logger.info("Admin configuration dropped the session; waiting for login.") + login_seen = re.search( + self._state_patterns[FOSCliState.PROVIDE_USERNAME.value], + self._admin_password_buffer.data, + ) + password_seen = re.search( + self._state_patterns[FOSCliState.PROVIDE_PASSWORD.value], + self._admin_password_buffer.data, + ) + self._admin_password_buffer.clear() + if login_seen: + return "login_prompt_seen" + if password_seen: + return "password_prompt_seen" + return "session_lost" + match = re.search(CURRENT_PASSWORD_PATTERN, self._admin_password_buffer.data) + if match: + self._admin_password_buffer.clear() + # We are still authenticated with the current credentials at this point. + self._terminal.wait_write(self._credentials.password, wait=None) + return "password_entered" + return None + + def _ready(self): + self._terminal.running = True + self._terminal.tn.close() + # calc startup time + startup_time = datetime.datetime.now() - self._start_time + self._logger.info(f"Startup complete in {startup_time}") diff --git a/fortinet/fortigate/docker/host_forwarded_bridge.py b/fortinet/fortigate/docker/host_forwarded_bridge.py new file mode 100644 index 00000000..b1d2e813 --- /dev/null +++ b/fortinet/fortigate/docker/host_forwarded_bridge.py @@ -0,0 +1,95 @@ +import logging + +import vrnetlab +from net_mgmt_strategy import NetMgmtStrategy + +BRIDGE_V4_ADDR = "172.31.255.29" +MGMT_V4_ADDR = "172.31.255.30" +V4_PREFIX_LENGTH = "30" +BRIDGE_V6_ADDR = "200::" +MGMT_V6_ADDR = "200::1" +V6_PREFIX_LENGTH = "127" + + +class HostForwardedBridge(NetMgmtStrategy): + mgmt_passthrough = False + + def configure_vm_mgmt(self, vm): + vm.mgmt_passthrough = False + vm.mgmt_address_ipv4 = f"{MGMT_V4_ADDR}/{V4_PREFIX_LENGTH}" + vm.mgmt_gw_ipv4 = BRIDGE_V4_ADDR + vm.mgmt_address_ipv6 = f"{MGMT_V6_ADDR}/{V6_PREFIX_LENGTH}" + vm.mgmt_gw_ipv6 = BRIDGE_V6_ADDR + + def gen_mgmt_netdev(self, vm): + return "bridge,br=br-mgmt,id=p00" + + def prep(self): + vrnetlab.run_command(["pkill", "socat"]) + + # redirecting incoming tcp traffic (except serial port 5000) from eth0 to management interface + vrnetlab.run_command( + f"iptables-nft -t nat -A PREROUTING -i eth0 -p tcp ! --dport 5000 -j DNAT --to-destination {MGMT_V4_ADDR}".split() + ) + vrnetlab.run_command( + f"ip6tables-nft -t nat -A PREROUTING -i eth0 -p tcp ! --dport 5000 -j DNAT --to-destination {MGMT_V6_ADDR}".split() + ) + # same redirection but for UDP + vrnetlab.run_command( + f"iptables-nft -t nat -A PREROUTING -i eth0 -p udp -j DNAT --to-destination {MGMT_V4_ADDR}".split() + ) + vrnetlab.run_command( + f"ip6tables-nft -t nat -A PREROUTING -i eth0 -p udp -j DNAT --to-destination {MGMT_V6_ADDR}".split() + ) + # masquerading the incoming traffic so SR OS is able to reply back + vrnetlab.run_command( + "iptables-nft -t nat -A POSTROUTING -o br-mgmt -j MASQUERADE".split() + ) + vrnetlab.run_command( + "ip6tables-nft -t nat -A POSTROUTING -o br-mgmt -j MASQUERADE".split() + ) + # allow sros breakout to management network by NATing via eth0 + vrnetlab.run_command( + "iptables-nft -t nat -A POSTROUTING -o eth0 -j MASQUERADE".split() + ) + vrnetlab.run_command( + "ip6tables-nft -t nat -A POSTROUTING -o eth0 -j MASQUERADE".split() + ) + + # ===================================================== + # set up bridge for management interface to a localhost + logger = logging.getLogger() + logger.info("Creating br-mgmt bridge for management interface") + # This is to whitlist all bridges + vrnetlab.run_command(["mkdir", "-p", "/etc/qemu"]) + vrnetlab.run_command(["echo 'allow all' > /etc/qemu/bridge.conf"], shell=True) + # Enable IPv6 inside the container + vrnetlab.run_command(["sysctl net.ipv6.conf.all.disable_ipv6=0"], shell=True) + # Enable IPv6 routing inside the container + vrnetlab.run_command(["sysctl net.ipv6.conf.all.forwarding=1"], shell=True) + vrnetlab.run_command(["brctl", "addbr", "br-mgmt"]) + vrnetlab.run_command( + ["echo 16384 > /sys/class/net/br-mgmt/bridge/group_fwd_mask"], + shell=True, + ) + vrnetlab.run_command(["ip", "link", "set", "br-mgmt", "up"]) + vrnetlab.run_command( + [ + "ip", + "addr", + "add", + "dev", + "br-mgmt", + f"{BRIDGE_V4_ADDR}/{V4_PREFIX_LENGTH}", + ] + ) + vrnetlab.run_command( + [ + "ip", + "addr", + "add", + "dev", + "br-mgmt", + f"{BRIDGE_V6_ADDR}/{V6_PREFIX_LENGTH}", + ] + ) diff --git a/fortinet/fortigate/docker/launch.py b/fortinet/fortigate/docker/launch.py index f420e2d6..50426c2f 100755 --- a/fortinet/fortigate/docker/launch.py +++ b/fortinet/fortigate/docker/launch.py @@ -1,13 +1,18 @@ #!/usr/bin/env python3 -import datetime import logging import os import re import signal import sys +import telnetlib import uuid import vrnetlab +from fos_cli_driver import FOSCliDriver +from host_forwarded_bridge import HostForwardedBridge +from net_mgmt_strategy import NetMgmtStrategy +from passthrough_redirect import PassthroughRedirect +from tftp import TFTPServer def handle_SIGCHLD(_unused_signal, _unused_frame): @@ -33,14 +38,21 @@ def trace(self, message, *args, **kws): logging.Logger.trace = trace +MGMT_PASSTHROUGH_DEFAULT = True +TFTP_PORT = 69 +TFTP_TID_RANGE = (52400, 52500) +TFTP_DIRECTORY = "/tftpboot" +GET_CONFIG_TRIGGER_FILE = "/get-config" class FortiOS_vm(vrnetlab.VM): - def __init__(self, hostname, username, password, conn_mode): + def __init__(self, hostname: str, username, password, conn_mode, mgmt_net: NetMgmtStrategy): + disk_image = None for e in os.listdir("."): if re.search(".qcow2$", e): disk_image = "./" + e - # call parents __init__ function here + if disk_image is None: + raise RuntimeError("Could not find image to boot") super(FortiOS_vm, self).__init__( username, password, @@ -49,7 +61,10 @@ def __init__(self, hostname, username, password, conn_mode): driveif="virtio", # fortios fails to respond to network requests if the pci bus is setup :D provision_pci_bus=False, + mgmt_passthrough=mgmt_net.mgmt_passthrough ) + + self.logger.info(f"Launching. commandline: {' '.join(sys.argv)}") self.conn_mode = conn_mode self.hostname = hostname self.num_nics = 12 @@ -57,96 +72,106 @@ def __init__(self, hostname, username, password, conn_mode): self.highest_port = 0 self.qemu_args.extend(["-uuid", os.getenv("FORTIGATE_UUID") or str(uuid.uuid4())]) self.spins = 0 - self.running = None - + self.stopped = False + self.waiting_for = False + self._mgmt_net = mgmt_net + self._mgmt_net.configure_vm_mgmt(self) + self.driver = FOSCliDriver(terminal=self, + mgmt_passthrough=self.mgmt_passthrough, + username=username, + password=password, + logger=logger, + mgmt_address_ipv4=self.mgmt_address_ipv4, + mgmt_gw_ipv4=self.mgmt_gw_ipv4, + mgmt_address_ipv6=self.mgmt_address_ipv6, + mgmt_gw_ipv6=self.mgmt_gw_ipv6, + hostname=hostname + ) # set up the extra empty disk image # for fortigate logs vrnetlab.run_command( ["qemu-img", "create", "-f", "qcow2", "empty.qcow2", "30G"] ) - self.qemu_args.extend( - [ - "-drive", - "if=virtio,format=qcow2,file=empty.qcow2,index=1", - ] - ) - + # Comma-separated list of disk sizes to install in the machine. as accepted by qemu-img create. + disk_specs = os.getenv("FOS_DISK_SPECS", "").split(",") + index = 1 + if disk_specs[0] == '': + self.logger.warn( + "No additional disks configured. Use FOS_DISK_SPECS to specify a comma-separated list of disk sizes") + return + index = 0 + for spec in disk_specs: + index += 1 + # set up the extra empty disk image + # for fortigate logs + vrnetlab.run_command( + ["qemu-img", "create", "-f", "qcow2", f"empty{index}.qcow2", spec] + ) + + self.qemu_args.extend( + [ + "-drive", + f"if=virtio,format=qcow2,file=empty{index}.qcow2,index={index}", + ] + ) + + index += 1 def bootstrap_spin(self): """This function should be called periodically to do work. returns False when it has failed and given up, otherwise True """ - if self.spins > 300: - # too many spins with no result -> restart - self.logger.warning("no output from serial console, restarting VCP") - self.stop() - self.start() - self.spins = 0 - return + self.driver.process_state() - (ridx, match, res) = self.tn.expect([b"login:", b"FortiGate-VM64-KVM #"], 1) - if match: # got a match! - if ridx == 0: # matched login prompt, so should login - self.logger.debug("ridx == 0") - self.logger.info("matched login prompt") - - self.wait_write(self.username, wait=None) - self.wait_write("", wait=self.username) - self.wait_write(self.password, wait="Password") - self.wait_write(self.password, wait=None) - - if ridx == 1: - # if we dont match the FortiGate-VM64-KVM # we assume we already have some configuration and - # may continue with configure the system to our needs. - self.logger.debug("ridx == 1") - self.wait_write("config system global", wait=None) - hostname_command = "set hostname " + self.hostname - self.wait_write(hostname_command, wait="global") - self.wait_write("end", wait=hostname_command) - self.running = True - self.tn.close() - # calc startup time - startup_time = datetime.datetime.now() - self.start_time - self.logger.info(f"Startup complete in { startup_time }") - return + def work(self): + super().work() + if self.running: + self._handle_get_config_trigger() - else: - # no match, if we saw some output from the router it's probably - # booting, so let's give it some more time - if res != b"": - self.logger.trace(f"OUTPUT FORTIGATE: {res.decode()}") - # reset spins if we saw some output - self.spins = 0 + def _handle_get_config_trigger(self): + if not os.path.exists(GET_CONFIG_TRIGGER_FILE): + return + try: + os.remove(GET_CONFIG_TRIGGER_FILE) + except Exception as exc: + self.logger.error(f"Failed to cleanup {GET_CONFIG_TRIGGER_FILE}: {exc}") + + self._telnet_connect() + self.driver.save_config() + self.logger.debug("get-config") + + def _telnet_connect(self): + try: + if self.tn is not None and self.tn.get_socket().fileno() >= 0: + self.tn.write(b"\r") + return + except Exception: + pass + try: + self.tn = telnetlib.Telnet("127.0.0.1", 5000 + self.num) + except Exception: + self.logger.exception("Failed to connect to VM serial port.") + self.stop() + raise + self.stopped = False + self.tn.write(b"\r") - self.spins += 1 + def gen_mgmt(self): + return self._mgmt_net.gen_mgmt(self) - def _wait_reset(self): - """ - This function waits for the login prompt after the VM was resetted. - If commands are issued that enforce a reboot this comes in hand. - e.g factoryreset or factoryreset2 - """ - self.logger.debug("waiting for reset") - wait_spins = 0 - while wait_spins < 90: - _, match, data = self.tn.expect([b"login: "], timeout=10) - self.logger.trace(data.decode("UTF-8")) - if match: - self.logger.debug("reset finished") - return True - wait_spins += 1 - self.logger.error("Reset took to long") - return False + def stop(self): + self.stopped = True + super().stop() class FortiOS(vrnetlab.VR): - def __init__(self, hostname, username, password, conn_mode): - super(FortiOS, self).__init__(username, password) + def __init__(self, hostname, username, password, conn_mode, mgmt_net: NetMgmtStrategy): + super(FortiOS, self).__init__(username, password, mgmt_passthrough=mgmt_net.mgmt_passthrough) self.logger.debug("Hostname") self.logger.debug(hostname) - self.vms = [FortiOS_vm(hostname, username, password, conn_mode)] + self.vms = [FortiOS_vm(hostname, username, password, conn_mode, mgmt_net=mgmt_net)] if __name__ == "__main__": @@ -157,8 +182,8 @@ def __init__(self, hostname, username, password, conn_mode): "--trace", action="store_true", help="enable trace level logging" ) parser.add_argument("--hostname", default="vr-fortinet", help="Fortinet hostname") - parser.add_argument("--username", default="admin", help="Username") - parser.add_argument("--password", default="admin", help="Password") + parser.add_argument("--username", default=None, help="Username") + parser.add_argument("--password", default=None, help="Password", nargs="?") parser.add_argument( "--connection-mode", default="tc", @@ -173,8 +198,32 @@ def __init__(self, hostname, username, password, conn_mode): logger.setLevel(logging.DEBUG) if args.trace: logger.setLevel(1) - vrnetlab.boot_delay() + + mgmt_passthrough = ( + os.environ.get("CLAB_MGMT_PASSTHROUGH", "").lower() == "true" + if os.environ.get("CLAB_MGMT_PASSTHROUGH") + else MGMT_PASSTHROUGH_DEFAULT + ) + + tftp_server = None + mgmt_net = None + if mgmt_passthrough: + mgmt_net = PassthroughRedirect([f"udp:{TFTP_PORT}", f"udp:{'-'.join(map(str, TFTP_TID_RANGE))}"]) + mgmt_net.prep() + tftp_server = TFTPServer(mgmt_net=mgmt_net, + srv_port=TFTP_PORT, + tid_range=TFTP_TID_RANGE, + directory=TFTP_DIRECTORY) + else: + mgmt_net = HostForwardedBridge() + mgmt_net.prep() + tftp_server = TFTPServer(srv_port=TFTP_PORT, + mgmt_net=mgmt_net, + directory=TFTP_DIRECTORY) + vr = FortiOS( - args.hostname, args.username, args.password, conn_mode=args.connection_mode + args.hostname, args.username, args.password, conn_mode=args.connection_mode, mgmt_net=mgmt_net ) + tftp_server.launch() + vrnetlab.boot_delay() vr.start() diff --git a/fortinet/fortigate/docker/net_mgmt_strategy.py b/fortinet/fortigate/docker/net_mgmt_strategy.py new file mode 100644 index 00000000..16be533c --- /dev/null +++ b/fortinet/fortigate/docker/net_mgmt_strategy.py @@ -0,0 +1,48 @@ +import vrnetlab + + +class NetMgmtStrategy: + mgmt_passthrough = False + + def prep(self): + pass + + def configure_vm_mgmt(self, vm: vrnetlab.VM): + """Copy strategy-owned addressing onto a vrnetlab VM.""" + return + + def gen_mgmt(self, vm: vrnetlab.VM): + if vm.mgmt_host_ip + 1 >= vm.mgmt_guest_ip: + vm.logger.error( + "Guest IP (%s) must be at least 2 higher than host IP(%s)", + vm.mgmt_guest_ip, + vm.mgmt_host_ip, + ) + + if ( + vm.snapshot_metadata + and "mac_addresses" in vm.snapshot_metadata + and len(vm.snapshot_metadata["mac_addresses"]) > 0 + ): + vm.mgmt_mac = vm.snapshot_metadata["mac_addresses"][0] + vm.logger.info(f"Using saved management MAC: {vm.mgmt_mac}") + else: + vm.mgmt_mac = vm.get_mgmt_mac() + + self.before_gen_mgmt(vm) + + return [ + "-device", + f"{vm.nic_type},netdev=p00,mac={vm.mgmt_mac}", + "-netdev", + self.gen_mgmt_netdev(vm), + ] + + def before_gen_mgmt(self, vm: vrnetlab.VM): + return + + def gen_mgmt_netdev(self, vm: vrnetlab.VM): + pass + + def write_ifup_script(self, vm: vrnetlab.VM): + return diff --git a/fortinet/fortigate/docker/passthrough_redirect.py b/fortinet/fortigate/docker/passthrough_redirect.py new file mode 100644 index 00000000..e0993202 --- /dev/null +++ b/fortinet/fortigate/docker/passthrough_redirect.py @@ -0,0 +1,201 @@ +import logging +import os + +import vrnetlab +from net_mgmt_strategy import NetMgmtStrategy + +DEFAULT_VETH_MAC_ADDR = "3a:3a:3a:3a:3a:3a" +DEFAULT_NS_NAME = "fakehost" +DEFAULT_ROOT_NS_VETH_LINK_NAME = "RA" +DEFAULT_PRIV_NS_VETH_LINK_NAME = "FA" +DEFAULT_TEMP_REDIR_DST = "169.254.254.254/16" + + +class PassthroughRedirect(NetMgmtStrategy): + mgmt_passthrough = True + + def __init__(self, + target_port_ranges, + logger: logging.Logger = None, + ns_name=DEFAULT_NS_NAME, + veth_mac_addr=DEFAULT_VETH_MAC_ADDR, + veth_root_ns_link_name=DEFAULT_ROOT_NS_VETH_LINK_NAME, + veth_priv_ns_link_name=DEFAULT_PRIV_NS_VETH_LINK_NAME, + temp_redir_dst=DEFAULT_TEMP_REDIR_DST): + """ + + :param target_port_ranges: List of strings with format: ":". + I.E. "udp:69", "tcp:52000-52400" + :param logger: + :param ns_name: + :param veth_mac_addr: + :param veth_root_ns_link_name: + :param veth_priv_ns_link_name: + :param temp_redir_dst: + """ + super().__init__() + self._target_port_ranges = target_port_ranges + if logger is None: + logger = logging.getLogger() + self.ns_name = ns_name + self._logger = logger + self._temp_redir_dst = temp_redir_dst + self._veth_priv_ns_link_name = veth_priv_ns_link_name + self._veth_root_ns_link_name = veth_root_ns_link_name + self._veth_mac_addr = veth_mac_addr + + def _cleanup_fakehost(self): + self._logger.info("Pre-clean of passthrough redirect") + cmds = [ + f"ip link del {self._veth_priv_ns_link_name}", + f"ip link del {self._veth_root_ns_link_name}", + f"ip netns del {self.ns_name}", + f"umount /run/netns/{self.ns_name}", + f"rm -f /run/netns/{self.ns_name} /var/run/netns/{self.ns_name}", + ] + + for cmd in cmds: + vrnetlab.run_command(cmd.split()) + + def prep(self): + self._cleanup_fakehost() + self._logger.info("Installing Mgmt Passthrough network redirect towards container") + # In management pass-through mode the container runs a tftp server in a dedicated namepace. + # This namespace will use the IPv4 default gateway of the container as interface + # tc flower rules will intercept tftp traffic and redirect it to this namespace + # create namespace + + vrnetlab.run_command(f"ip netns add {self.ns_name}".split()) + # create vethts: FA in fakehost ns, RA in "root" ns + vrnetlab.run_command( + f"ip link add {self._veth_priv_ns_link_name} type veth peer name {self._veth_root_ns_link_name}".split()) + # assign FA veth to ns + vrnetlab.run_command(f"ip link set {self._veth_priv_ns_link_name} netns {self.ns_name}".split()) + # enable veth root ns + vrnetlab.run_command(f"ip link set {self._veth_root_ns_link_name} up".split()) + # enable loop in ns + vrnetlab.run_command(f"ip netns exec {self.ns_name} ip link set dev lo up".split()) + # enable veth in fakehost ns + vrnetlab.run_command(f"ip netns exec {self.ns_name} ip link set {self._veth_priv_ns_link_name} up".split()) + # assign a dummy mac that will not collide with the real docker bridge mac address + vrnetlab.run_command( + f"ip netns exec {self.ns_name} ip link set dev {self._veth_priv_ns_link_name} address {DEFAULT_VETH_MAC_ADDR}".split() + ) + # configure a temporary ip address so the tftp server can start. + # modified later in the startup process in the create_tc_tap_mgmt_ifup function + vrnetlab.run_command( + f"ip netns exec {self.ns_name} ip addr add {self._temp_redir_dst} dev {self._veth_priv_ns_link_name}".split() + ) + # block arp responses in fakehost namespace so it doesn't interfere with root namespace + vrnetlab.run_command( + f"ip netns exec {self.ns_name} sysctl -w net.ipv4.conf.all.arp_ignore=8".split() + ) + + def before_gen_mgmt(self, vm): + self.write_ifup_script(vm) + + def gen_mgmt_netdev(self, vm): + return "tap,ifname=tap0,id=p00,script=/etc/tc-tap-mgmt-ifup,downscript=no" + + def write_ifup_script(self, vm): + mgmt_ip_v4_address, mgmt_ip_v4_prefixlen = vm.mgmt_address_ipv4.split("/") + ifup_script = self.get_tc_tap_mgmt_ifup( + vm.mgmt_gw_ipv4, + mgmt_ip_v4_prefixlen, + mgmt_ip_v4_address, + vm.mgmt_mac, + ) + + with open("/etc/tc-tap-mgmt-ifup", "w") as f: + f.write(ifup_script) + os.chmod("/etc/tc-tap-mgmt-ifup", 0o777) + + def get_tc_tap_mgmt_ifup(self, + redir_addr, + redir_prefix_len, + src_addr, + src_mac): + + # override the parent's function with sros requirements + # this is used when using pass-through mode for mgmt connectivity + """Create tap ifup script that is used in tc datapath mode, specifically for the management interface""" + ifup_script = """#!/bin/bash + + ip link set tap0 up + ip link set tap0 mtu 65000 + + # disable IPv6 to avoid sending periodic traffic like router solicitations from the vrnetlab container + ip -6 addr flush tap0 + + # create tc eth<->tap redirect rules + + tc qdisc add dev eth0 clsact + + # exception for TCP ports 5000-5007 + tc filter add dev eth0 ingress prio 1 protocol ip flower ip_proto tcp dst_port 5000-5007 action pass + + # mirror ARP traffic to container + tc filter add dev eth0 ingress prio 2 protocol arp flower action mirred egress mirror dev tap0 + # redirect rest of ingress traffic of eth0 to egress of tap0 + tc filter add dev eth0 ingress prio 3 flower action mirred egress redirect dev tap0 + + tc qdisc add dev tap0 clsact + # redirect all ingress traffic of tap0 to egress of eth0 + tc filter add dev tap0 ingress flower action mirred egress redirect dev eth0 + + # clone management MAC of the VM + ip link set dev eth0 address {SRC_MAC} + + tc qdisc add dev {RA} clsact + + # configure the ip address of the namespace as it was the host and remove the temporary one + ip netns exec {NS_NAME} ip addr add {REDIR_ADDR}/{REDIR_PREFIX_LEN} dev {FA} + ip netns exec {NS_NAME} ip addr del {TEMP_REDIR_DST} dev {FA} + + """ + prio = 1 + for entry in self._target_port_ranges: + proto, port = entry.split(":") + ifup_script += (""" + # Redirect traffic from VM to private NS + tc filter add dev tap0 ingress protocol ip prio {PRIO} \ + flower ip_proto {PROTO} dst_port {PORT} dst_ip {REDIR_ADDR} \ + action pedit ex munge eth dst set {PRIV_NS_VETH_MAC_ADDR} pipe \ + action mirred egress redirect dev {RA} + + """ + .replace("{PROTO}", proto) + .replace("{PORT}", port) + .replace("{PRIO}", str(prio))) + prio += 1 + + for entry in self._target_port_ranges: + proto, port = entry.split(":") + ifup_script += (""" + # Redirect traffic from private NS to VM + tc filter add dev {RA} ingress protocol ip prio {PRIO} \ + flower ip_proto {PROTO} src_port {PORT} dst_ip {SRC_ADDR} \ + action pedit ex munge eth dst set {SRC_MAC} pipe \ + action mirred egress redirect dev tap0 + + """ + .replace("{PROTO}", proto) + .replace("{PORT}", port) + .replace("{PRIO}", str(prio))) + prio += 1 + + # FA, RA, TEMP_REDIR_DST + ifup_script = ifup_script.replace("{RA}", self._veth_root_ns_link_name) + ifup_script = ifup_script.replace("{FA}", self._veth_priv_ns_link_name) + ifup_script = ifup_script.replace("{NS_NAME}", self.ns_name) + ifup_script = ifup_script.replace("{TEMP_REDIR_DST}", self._temp_redir_dst) + ifup_script = ifup_script.replace("{SRC_MAC}", src_mac) + ifup_script = ifup_script.replace( + "{PRIV_NS_VETH_MAC_ADDR}", self._veth_mac_addr + ) + ifup_script = ifup_script.replace("{REDIR_ADDR}", redir_addr) + ifup_script = ifup_script.replace("{REDIR_PREFIX_LEN}", redir_prefix_len) + ifup_script = ifup_script.replace("{SRC_ADDR}", src_addr) + self._logger.info(f"Traffic towards {redir_addr} on ports: [] redirected towards {self.ns_name} mac: " + f"{self._veth_mac_addr} and return traffic directed to IP {src_addr} with MAC: {src_mac}") + return ifup_script diff --git a/fortinet/fortigate/docker/tftp.py b/fortinet/fortigate/docker/tftp.py new file mode 100644 index 00000000..c36a4c1f --- /dev/null +++ b/fortinet/fortigate/docker/tftp.py @@ -0,0 +1,88 @@ +# In pass-through mode, we also spin up a tftp server, but in this case we create a new namespace +# inside the container that simulates the IP addressing of the host. +# we redirect traffic to this ns by using tc flower filters +import logging +import typing +from abc import abstractmethod, ABCMeta + +import vrnetlab + + +class _TFTPLauncher(metaclass=ABCMeta): + + def __init__(self, addr, srv_port, directory, tid_range): + super().__init__() + self._directory = directory + self._srv_port = srv_port + self._tid_range = tid_range + self._addr = addr + + @abstractmethod + def launch(self): ... + + +class _HostForwardedLauncher(_TFTPLauncher): + def launch(self): + logger = logging.getLogger() + logger.info(f"Launching TFTP Server in Host-Forwarded mode. at={self._addr}:{self._srv_port}") + cmd = [ + "in.tftpd", + "--listen", + "--user", + "root", + "-a", + f"{self._addr}:{self._srv_port}", + "-s", + "-c", + "-v", + "-p", + ] + if self._tid_range is not None: + cmd.append("-R") + cmd.append(":".join(map(str, self._tid_range))) + cmd.append(self._directory) + vrnetlab.run_command(cmd) + + +class _PassthroughLauncher(_TFTPLauncher): + + def launch(self, ): + logger = logging.getLogger() + logger.info(f"Launching TFTP Server in Passthrough mode. at={self._addr}:{self._srv_port}") + + # start tftp in ns, assign ports to server so it's easier to track it with flower filters + vrnetlab.run_command( + [ + "ip", + "netns", + "exec", + "fakehost", + "in.tftpd", + "--listen", + "--user", + "root", + "-a", + f"{self._addr}:{self._srv_port}", + "-R", + ":".join(map(str, self._tid_range)), + "-s", + "-c", + "-v", + "-p", + self._directory, + ] + ) + + +class TFTPServer(_TFTPLauncher): + def __init__(self, addr="0.0.0.0", directory="/tftpboot", mgmt_net=None, srv_port=69, + tid_range: typing.Iterable[int] = None): + super().__init__(addr, srv_port, directory, tid_range) + if mgmt_net and mgmt_net.mgmt_passthrough: + self.launcher = _PassthroughLauncher(addr, srv_port, directory, tid_range) + else: + self.launcher = _HostForwardedLauncher(addr, srv_port, directory, tid_range) + + def launch(self): + self.launcher.launch() + vrnetlab.run_command(["chmod", "-R", "777", self._directory])