Python调用AWS API日志无法记录执行Linux用户问题求助

问题描述 #

现有Python脚本可正常调用AWS API添加安全组入站规则，但无法将执行脚本的Linux用户信息写入日志文件。

问题原因 #

原代码中虽然通过extra参数传递了用户信息，但日志格式字符串未包含%(user)s占位符，导致logging模块无法解析并输出用户信息。

修复方案 #

1. 调整日志格式配置 #

在logging.basicConfig的format参数中添加%(user)s占位符，确保用户信息能被输出到日志。

2. 简化extra参数使用 #

无需依赖defaultdict，直接传递包含user键的字典即可；同时增加用户获取失败的异常处理，避免脚本崩溃。

修改后的完整代码 #

import os
import sys
import boto3
import logging
import getpass
import argparse

# 配置日志 - 添加%(user)s占位符
logging.basicConfig(
    filename='/opt/tango/aws/awsapiBUILD_RULE_in_SG.log',
    format='%(asctime)s - %(levelname)s - %(user)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
    level=logging.INFO
)

def create_session():
    access_key = os.environ.get('AWS_ACCESS_KEY_ID')
    secret_key = os.environ.get('AWS_SECRET_ACCESS_KEY')
    session = boto3.Session(
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        region_name='us-east-1'
    )
    return session

def create_sg_rule(group_id, ip_protocol, from_port, to_port, cidr_ip, description):
    ec2 = boto3.client('ec2', region_name='us-east-1')
    # 获取当前用户，失败时设为默认值
    try:
        user = getpass.getuser()
    except Exception:
        user = 'unknown'

    try:
        response = ec2.authorize_security_group_ingress(
            GroupId=group_id,
            IpPermissions=[
                {
                    'IpProtocol': ip_protocol,
                    'FromPort': from_port,
                    'ToPort': to_port,
                    'IpRanges': [
                        {
                            'CidrIp': cidr_ip,
                            'Description': description
                        },
                    ],
                },
            ]
        )
        print(f'Ingress rule added to security group {group_id}')
        logging.info(
            f'Ingress rule added to security group {group_id}',
            extra={'user': user}
        )
        output = {"success": True}
    except Exception as e:
        error_message = f'Error: {e}'
        print(f'{error_message}, User: {user}')
        logging.error(
            error_message,
            extra={'user': user}
        )
        output = {"error": str(e)}
    return output

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Add ingress rule to AWS security group.')
    parser.add_argument('--group-id', type=str, required=True, help='The ID of the security group to modify')
    parser.add_argument('--ip-protocol', type=str, required=True, help='The IP protocol for the ingress rule')
    parser.add_argument('--from-port', type=int, required=True, help='The starting port for the ingress rule')
    parser.add_argument('--to-port', type=int, required=True, help='The ending port for the ingress rule')
    parser.add_argument('--cidr-ip', type=str, required=True, help='The CIDR IP range to allow access from')
    parser.add_argument('--description', type=str, required=True, help='The description of the ingress rule')
    args = parser.parse_args()

    session = create_session()
    create_sg_rule(args.group_id, args.ip_protocol, args.from_port, args.to_port, args.cidr_ip, args.description)

替代方案（无需修改日志格式） #

如果不想调整日志格式，可直接将用户信息拼接到日志消息中，示例：

logging.info(f'User {user} added ingress rule to security group {group_id}')

这种方式无需依赖extra参数，实现更直接。

内容的提问来源于stack exchange，提问作者ilgtech