Skip to content

[BUG] Strip ANSI escape codes from Terraform/Ansible output in API and logs (fixes #1366, fixes #1367) #1401

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cray44 wants to merge 1 commit into
base: develop
Choose a base branch
from
+64 −31
Open

[BUG] Strip ANSI escape codes from Terraform/Ansible output in API and logs (fixes #1366, fixes #1367) #1401

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 12 additions & 11 deletions api/app.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
from typing import Dict, Any, Optional, Tuple, List

from flask import jsonify, request
from attack_range.utils import strip_ansi
from flask_openapi3 import OpenAPI, Info, Tag
from flask_cors import CORS
from pydantic import ValidationError
Expand Down Expand Up @@ -490,16 +491,16 @@ def run_build_vpn_phase(config: Dict[str, Any], config_path: str, attack_range_i
with operations_lock:
running_operations[attack_range_id]["status"] = "error"
running_operations[attack_range_id]["end_time"] = datetime.now().isoformat()
running_operations[attack_range_id]["error"] = str(e)
running_operations[attack_range_id]["error"] = strip_ansi(str(e))
running_operations[attack_range_id]["error_phase"] = "build_vpn"
running_operations[attack_range_id]["traceback"] = traceback.format_exc()
running_operations[attack_range_id]["traceback"] = strip_ansi(traceback.format_exc())

# Update error status in config file (via controller if available, else direct write)
try:
controller = AttackRangeController(config, config_path=config_path)
controller.config_manager.update_status("error", error=str(e), error_phase="build_vpn")
controller.config_manager.update_status("error", error=strip_ansi(str(e)), error_phase="build_vpn")
except Exception:
_write_config_error_status(config_path, str(e), "build_vpn")
_write_config_error_status(config_path, strip_ansi(str(e)), "build_vpn")


def run_build_lab_phase(attack_range_id: str):
Expand Down Expand Up @@ -549,18 +550,18 @@ def run_build_lab_phase(attack_range_id: str):
config_path = get_config_path_from_attack_range_id(attack_range_id)
running_operations[attack_range_id]["status"] = "error"
running_operations[attack_range_id]["end_time"] = datetime.now().isoformat()
running_operations[attack_range_id]["error"] = str(e)
running_operations[attack_range_id]["error"] = strip_ansi(str(e))
running_operations[attack_range_id]["error_phase"] = "build_lab"
running_operations[attack_range_id]["traceback"] = traceback.format_exc()
running_operations[attack_range_id]["traceback"] = strip_ansi(traceback.format_exc())

# Update error status in config file (via controller if available, else direct write)
if config_path:
try:
config = load_yaml_file(config_path)
controller = AttackRangeController(config, config_path=config_path)
controller.config_manager.update_status("error", error=str(e), error_phase="build_lab")
controller.config_manager.update_status("error", error=strip_ansi(str(e)), error_phase="build_lab")
except Exception:
_write_config_error_status(config_path, str(e), "build_lab")
_write_config_error_status(config_path, strip_ansi(str(e)), "build_lab")


def run_destroy_operation(
Expand Down Expand Up @@ -600,13 +601,13 @@ def run_destroy_operation(
with operations_lock:
running_operations[attack_range_id]["status"] = "failed"
running_operations[attack_range_id]["end_time"] = datetime.now().isoformat()
running_operations[attack_range_id]["error"] = str(e)
running_operations[attack_range_id]["traceback"] = traceback.format_exc()
running_operations[attack_range_id]["error"] = strip_ansi(str(e))
running_operations[attack_range_id]["traceback"] = strip_ansi(traceback.format_exc())

if config_path:
try:
c = AttackRangeController(config, config_path=config_path)
c.config_manager.update_status("failed", error=str(e))
c.config_manager.update_status("failed", error=strip_ansi(str(e)))
except Exception:
pass

Expand Down
27 changes: 17 additions & 10 deletions attack_range/managers/ansible_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import logging
import ansible_runner
from typing import Optional, Dict, Any
from attack_range.utils import strip_ansi

# Galaxy role that must always be updated to latest before VPN playbooks (vpn.yaml, vpn_config.yaml)
WIREGUARD_GALAXY_ROLE = "p4t12ick.ar_wireguard_vpn"
Expand Down Expand Up @@ -598,11 +599,15 @@ def run_ansible_playbook(self, playbook_name: str, extra_vars: dict = None) -> N
extra_vars_parts = [f"-e {k}={shlex.quote(str(v))}" for k, v in extra_vars.items()]
cmdline = f"{cmdline} {' '.join(extra_vars_parts)}"

ansible_env = os.environ.copy()
ansible_env["ANSIBLE_NO_COLOR"] = "1"

runner = ansible_runner.run(
private_data_dir=self.ansible_dir,
cmdline=cmdline,
playbook=playbook_name,
verbosity=2, # Increased verbosity for better error diagnostics
verbosity=2,
envvars=ansible_env,
)

if extra_vars_path and os.path.exists(extra_vars_path):
Expand Down Expand Up @@ -633,22 +638,21 @@ def run_ansible_playbook(self, playbook_name: str, extra_vars: dict = None) -> N
if event.get('event') in ['runner_on_failed', 'runner_on_unreachable', 'runner_on_error']:
host = event_data.get('host', 'unknown')
task = event_data.get('task', 'unknown')
msg = event_data.get('msg', 'No error message')
msg = strip_ansi(str(event_data.get('msg', 'No error message')))
error_events.append(f"Host: {host}, Task: {task}, Error: {msg}")

if error_events:
self.logger.error("Ansible playbook errors:")
for error in error_events[-10:]: # Show last 10 errors
for error in error_events[-10:]:
self.logger.error(f" - {error}")

# Try to read stdout/stderr from the runner's artifact directory
try:
stdout_path = os.path.join(self.ansible_dir, 'artifacts', str(runner.config.ident), 'stdout')
if os.path.exists(stdout_path):
with open(stdout_path, 'r') as f:
stdout_content = f.read()
stdout_content = strip_ansi(f.read())
if stdout_content:
# Log last 50 lines of stdout
lines = stdout_content.strip().split('\n')
self.logger.error("Last 50 lines of Ansible output:")
for line in lines[-50:]:
Expand Down Expand Up @@ -721,11 +725,15 @@ def run_ansible_playbook_safe(self, playbook_name: str, extra_vars: dict = None)
extra_vars_parts = [f"-e {k}={shlex.quote(str(v))}" for k, v in extra_vars.items()]
cmdline = f"{cmdline} {' '.join(extra_vars_parts)}"

ansible_env = os.environ.copy()
ansible_env["ANSIBLE_NO_COLOR"] = "1"

runner = ansible_runner.run(
private_data_dir=self.ansible_dir,
cmdline=cmdline,
playbook=playbook_name,
verbosity=2, # Increased verbosity for better error diagnostics
verbosity=2,
envvars=ansible_env,
)

if extra_vars_path and os.path.exists(extra_vars_path):
Expand Down Expand Up @@ -970,23 +978,22 @@ def run_ansible_playbook_safe(self, playbook_name: str, extra_vars: dict = None)
if event.get('event') in ['runner_on_failed', 'runner_on_unreachable', 'runner_on_error']:
host = event_data.get('host', 'unknown')
task = event_data.get('task', 'unknown')
msg = event_data.get('msg', 'No error message')
msg = strip_ansi(str(event_data.get('msg', 'No error message')))
error_events.append(f"Host: {host}, Task: {task}, Error: {msg}")

if error_events:
error_details.extend(error_events)
self.logger.error("Ansible playbook errors:")
for error in error_events[-10:]: # Show last 10 errors
for error in error_events[-10:]:
self.logger.error(f" - {error}")

# Try to read stdout/stderr from the runner's artifact directory
try:
stdout_path = os.path.join(self.ansible_dir, 'artifacts', str(runner.config.ident), 'stdout')
if os.path.exists(stdout_path):
with open(stdout_path, 'r') as f:
stdout_content = f.read()
stdout_content = strip_ansi(f.read())
if stdout_content:
# Get last 50 lines of stdout
lines = stdout_content.strip().split('\n')
last_lines = lines[-50:] if len(lines) > 50 else lines
error_details.append("Last Ansible output:")
Expand Down
26 changes: 16 additions & 10 deletions attack_range/managers/terraform_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,8 @@
import subprocess
import shutil
import logging
from python_terraform import Terraform, IsNotFlagged
from python_terraform import Terraform, IsNotFlagged, IsFlagged
from attack_range.utils import strip_ansi


class TerraformManager:
Expand Down Expand Up @@ -110,7 +111,7 @@ def init(self, backend_was_created: bool = False) -> None:
text=True
)
if init_result.returncode != 0:
msg = init_result.stderr or init_result.stdout or "Terraform init failed"
msg = strip_ansi(init_result.stderr or init_result.stdout or "Terraform init failed")
self.logger.error(f"Terraform init failed: {msg}")
raise RuntimeError(msg)
self.logger.info("Terraform initialized successfully")
Expand All @@ -131,7 +132,7 @@ def init_for_destroy(self) -> None:
text=True
)
if init_result.returncode != 0:
msg = init_result.stderr or init_result.stdout or "Terraform init failed"
msg = strip_ansi(init_result.stderr or init_result.stdout or "Terraform init failed")
self.logger.error(f"Terraform init failed: {msg}")
raise RuntimeError(msg)
self.logger.info("Terraform initialized successfully")
Expand All @@ -144,9 +145,12 @@ def apply(self) -> None:
"""
self.logger.info("Applying terraform configuration...")
return_code, stdout, stderr = self.terraform.apply(
capture_output="yes", skip_plan=True, no_color=IsNotFlagged
capture_output="yes", skip_plan=True, no_color=IsFlagged
)

stdout = strip_ansi(stdout) if stdout else stdout
stderr = strip_ansi(stderr) if stderr else stderr

if return_code != 0:
self.logger.error("Terraform apply failed!")
if stderr:
Expand Down Expand Up @@ -189,31 +193,33 @@ def destroy(self) -> None:
self.logger.info("Destroying terraform infrastructure...")
return_code, stdout, stderr = self.terraform.destroy(
capture_output="yes",
no_color=IsNotFlagged,
no_color=IsFlagged,
force=IsNotFlagged,
auto_approve=True,
)

stdout = strip_ansi(stdout) if stdout else stdout
stderr = strip_ansi(stderr) if stderr else stderr

if return_code != 0:
# Check if error is due to locked state
error_output = stderr or stdout or ""
if "Error acquiring the state lock" in error_output or "lock is currently held" in error_output.lower():
# Try to extract lock ID from error message
import re
lock_id_match = re.search(r'Lock ID:\s*([a-f0-9-]+)', error_output, re.IGNORECASE)
if lock_id_match:
lock_id = lock_id_match.group(1)
self.logger.warning(f"Terraform state is locked. Attempting to unlock with ID: {lock_id}")
self.force_unlock(lock_id)
# Retry destroy after unlock
self.logger.info("Retrying terraform destroy after unlock...")
return_code, stdout, stderr = self.terraform.destroy(
capture_output="yes",
no_color=IsNotFlagged,
no_color=IsFlagged,
force=IsNotFlagged,
auto_approve=True,
)

stdout = strip_ansi(stdout) if stdout else stdout
stderr = strip_ansi(stderr) if stderr else stderr

if return_code != 0:
self.logger.error(f"Terraform destroy failed: {stderr}")
raise RuntimeError(stderr or "Terraform destroy failed")
Expand Down
19 changes: 19 additions & 0 deletions attack_range/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,30 @@
"""

import os
import re
import uuid
import yaml
from typing import Dict, Any, Tuple, Optional
from datetime import datetime

_ANSI_ESCAPE_RE = re.compile(r'\x1b\[[0-9;]*[a-zA-Z]')


def strip_ansi(text: str) -> str:
"""
Remove ANSI escape sequences from text.

Subprocess output from tools like Terraform and Ansible may contain
terminal color codes (e.g., '\\x1b[31m') that are not readable in
web app logs, structured error messages, or API responses.

:param text: String that may contain ANSI escape sequences
:return: String with all ANSI sequences removed
"""
if not text:
return text
return _ANSI_ESCAPE_RE.sub('', text)


def resolve_template_path(template: str, templates_dir: str) -> str:
"""
Expand Down