Skip to content
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

feat(module/asg): Update Lambda runtime to Python 3.11 #59

Merged
merged 2 commits into from
Jul 11, 2024
Merged

feat(module/asg): Update Lambda runtime to Python 3.11 #59

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7149d7c
Reformat code, remove target exceptions, adjust lines to 120 characters
sebastianczech Jul 11, 2024
1a52ddd
Use runtime python3.11
sebastianczech Jul 11, 2024
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
2 changes: 1 addition & 1 deletion modules/asg/main.tf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,7 +310,7 @@ resource "aws_lambda_function" "this" {
role = aws_iam_role.this.arn
handler = "lambda.lambda_handler"
source_code_hash = data.archive_file.this.output_base64sha256
runtime = "python3.8"
runtime = "python3.11"
timeout = var.lambda_timeout
reserved_concurrent_executions = var.reserved_concurrent_executions
tracing_config {
Expand Down
48 changes: 30 additions & 18 deletions modules/asg/scripts/lambda.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@
from xml.etree.ElementTree import Element

from boto3 import client
from botocore.exceptions import ClientError, ParamValidationError
from botocore.config import Config
from botocore.exceptions import ClientError, ParamValidationError

from panos.panorama import Panorama

Expand Down Expand Up @@ -290,11 +290,11 @@ def disable_source_dest_check(self, interface_id: str):
"""
self.logger.info(f"Disable source_dest_check for network interface {interface_id}")
self.ec2_client.modify_network_interface_attribute(
NetworkInterfaceId=interface_id,
SourceDestCheck={
'Value': False,
}
)
NetworkInterfaceId=interface_id,
SourceDestCheck={
'Value': False,
}
)

def modify_network_interface(self, interface_id: str, attachment_id: str, source_dest_check: bool = True):
"""
Expand Down Expand Up @@ -365,6 +365,7 @@ def ip_network_interface(self, instance_id: str, device_index: str):
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2/client/describe_network_interfaces.html

:param instance_id: EC2 Instance id
:param device_index: Network interface index
:return: none
"""
description = self.ec2_client.describe_network_interfaces(
Expand All @@ -381,7 +382,7 @@ def ip_network_interface(self, instance_id: str, device_index: str):
)
try:
return description['NetworkInterfaces'][0]['PrivateIpAddress']
except IndexError as e:
except IndexError:
return None

def register_untrust_ip_as_target(self, ip_target_groups: list, untrust_ip: str):
Expand All @@ -406,7 +407,7 @@ def register_untrust_ip_as_target(self, ip_target_groups: list, untrust_ip: str)
},
]
)
except ParamValidationError as e:
except ParamValidationError:
self.logger.error(f"Unable to register target with IP {untrust_ip}")

def deregister_untrust_ip_as_target(self, ip_target_groups: list, untrust_ip: str):
Expand All @@ -430,7 +431,7 @@ def deregister_untrust_ip_as_target(self, ip_target_groups: list, untrust_ip: st
},
]
)
except ParamValidationError as e:
except ParamValidationError:
self.logger.error(f"Unable to deregister target with IP {untrust_ip}")

def panorama_cmd(self, panorama, cmd: str, xml: bool = True, cmd_xml: bool = True) -> Element:
Expand All @@ -454,8 +455,8 @@ def check_ssm_param(self, ssm_param_name: str) -> dict:
:ssm_param_name: Parameter name
:return: dict
"""
ssm_param_list = self.ssm_client.get_parameter(Name=ssm_param_name, WithDecryption=True).get("Parameter").get("Value"). \
replace("\'", "\"")
ssm_param_list = self.ssm_client.get_parameter(Name=ssm_param_name, WithDecryption=True). \
get("Parameter").get("Value").replace("\'", "\"")
return loads(ssm_param_list)

def delicense_fw(self, instance_id) -> bool:
Expand Down Expand Up @@ -489,13 +490,19 @@ def delicense_fw(self, instance_id) -> bool:
# Check if first Panorama is active - if not, the use second Panorama for de-licensing
if self.check_is_active_in_ha(panorama_hostname, panorama_username, panorama_password):
# De-license using active, first Panorama instance from Active-Passive HA cluster
delicensed = self.request_panorama_delicense_fw(vmseries_ip_address, panorama_hostname, panorama_username, panorama_password, panorama_lm_name)
delicensed = self.request_panorama_delicense_fw(vmseries_ip_address, panorama_hostname,
panorama_username, panorama_password,
panorama_lm_name)
else:
# De-license using active, second Panorama instance from Active-Passive HA cluster
delicensed = self.request_panorama_delicense_fw(vmseries_ip_address, panorama_hostname2, panorama_username, panorama_password, panorama_lm_name)
delicensed = self.request_panorama_delicense_fw(vmseries_ip_address, panorama_hostname2,
panorama_username, panorama_password,
panorama_lm_name)
else:
# De-license using the only 1 Panorama instance
delicensed = self.request_panorama_delicense_fw(vmseries_ip_address, panorama_hostname, panorama_username, panorama_password, panorama_lm_name)
delicensed = self.request_panorama_delicense_fw(vmseries_ip_address, panorama_hostname,
panorama_username, panorama_password,
panorama_lm_name)

return delicensed
else:
Expand Down Expand Up @@ -537,22 +544,26 @@ def check_is_active_in_ha(self, panorama_hostname, panorama_username, panorama_p
self.logger.info(f"Error while checking high-availability state for Panorama {panorama_hostname}")
return False

def request_panorama_delicense_fw(self, vmseries_ip_address, panorama_hostname, panorama_username, panorama_password, panorama_lm_name) -> bool:
def request_panorama_delicense_fw(self, vmseries_ip_address, panorama_hostname, panorama_username,
panorama_password, panorama_lm_name) -> bool:
"""
Function used to de-license VM-Series using plugin sw_fw_license running on Panorama server

:param vmseries_ip_address: IP address of the MGMT interface for VM-Series
:param panorama_hostname: Hostname of the Panorama server
:param panorama_username: Account's name
:param panorama_password: Account's password
:param panorama_lm_name: License manager name (in plugin sw_fw_license)
:return: True if VM-Series was de-licensed correctly, False in other case
"""
try:
# Set status of delicensing
delicensed = False

# Connect to selected Panorama instance
self.logger.info(f"Connecting to '{panorama_hostname}' using user '{panorama_username}' to license manager '{panorama_lm_name}'")
self.logger.info(
f"Connecting to '{panorama_hostname}' using user '{panorama_username}' "
f"to license manager '{panorama_lm_name}'")
panorama = Panorama(hostname=panorama_hostname,
api_username=panorama_username,
api_password=panorama_password)
Expand All @@ -575,10 +586,11 @@ def request_panorama_delicense_fw(self, vmseries_ip_address, panorama_hostname,
serial_obj = fw.find("serial")
if serial_obj is not None:
serial = serial_obj.text
# If IP address is the same as destroyed VM and serial is not none, then delicense firewall
# If IP address is the same as destroyed VM and serial is not none, then delicense FW
if serial_obj.text is not None:
self.logger.info(f"De-licensing firewall: {serial} ...")
cmd = f"request plugins sw_fw_license deactivate license-manager \"{panorama_lm_name}\" devices member \"{serial}\""
cmd = (f"request plugins sw_fw_license deactivate license-manager "
f"\"{panorama_lm_name}\" devices member \"{serial}\"")
resp_parsed = self.panorama_cmd(panorama, cmd)
if resp_parsed.attrib["status"] == "success":
self.logger.info(f"De-licensing firewall: {serial} succeeded")
Expand Down
Loading