Added integration with TA

frameworks/helper/WATools.py

@@ -83,7 +83,7 @@ def createReportIfNotExists(self):
             'Description': 'Auto generated by ServiceScreener',
             'Environment': 'PRODUCTION',
             'AccountIds': [self.stsInfo['Account']],
-            'AwsRegions': self.region,
+            'AwsRegions': [self.region],
             'ReviewOwner': self.stsInfo['Arn'],
             'Lenses': [self.waInfo['LensesAlias']]
         }

services/PageBuilder.py

@@ -25,7 +25,8 @@ class PageBuilder:
         'rds': 'database',
         's3': 'hdd',
         'Modernize': 'chart-line',
-        'Finding': 'bug'
+        'Findings': 'bug',
+        'TA': 'user-md'
     }
 
     frameworkIcon = 'tasks'
@@ -201,20 +202,20 @@ def generateSummaryCardContent(self, summary):
         resHtml = []
         for region, resource in resources.items():
             items = []
-            resHtml.append(f"<dd>{region}: ")
+            resHtml.append(f"<dd class='detail-regions'>{region}: ")
             for identifier in resource:
                 items.append(f"<a href='#{self.service}-{identifier}'>{identifier}</a>")
             resHtml.append(" | ".join(items))
             resHtml.append("</dd>")
 
-        output.append("<dl><dt>Description</dt><dd>" + summary['^description'] + "</dd><dt>Resources</dt>" + "".join(resHtml))
+        output.append("<dl><dt>Description</dt><dd class='detail-desc'>" + summary['^description'] + "</dd><dt>Resources</dt>" + "".join(resHtml))
 
         hasTags = self.generateSummaryCardTag(summary)
         if len(hasTags.strip()) > 0:
             output.append(f"<dt>Label</dt><dd>{hasTags}</dd>")
 
         if summary['__links']:
-            output.append("<dt>Recommendation</dt><dd>" + "</dd><dd>".join(summary['__links']) + "</dd>")
+            output.append("<dt>Recommendation</dt><dd class='detail-href'>" + "</dd><dd class='detail-href''>".join(summary['__links']) + "</dd>")
 
         output.append("</dl>")
 
@@ -684,26 +685,34 @@ def genaiModalHtml(self):
 
 genaiResp = $('#genai-modal-response')
 $('#genai-savequery').click(function(){
+  sbtn = $(this)
   genaikeys = $('#genai-key').val().split('|')
 
   if((genaikeys.length < 2) || (genaikeys.length > 2)){
     alert('invalid keys')
     return
   }
 
+  sbtn.prop('disabled', true)
+  genaiResp.text("... generating results, it can times, please be patient ...")
+
   a_url = genaikeys[0]
   a_key = genaikeys[1]
 
+  myJsonData = {'api_data': currentInfo}
   $.ajax({
     url: a_url,
-    headers: {'x-api-key': a_key},
+    // headers: {'x-api-key': a_key},
     type: 'POST',
-    data: JSON.stringify(currentInfo),
+    data: JSON.stringify(myJsonData),
     contentType: 'application/json',
+    dataType: 'json',
     success: function(response) {
       genaiResp.text(response['createdAt'])
+      sbtn.prop('disabled', false)
     },
     error: function(xhr, status, error) {
+      sbtn.prop('disabled', false)    
       genaiResp.text("Error..., check console.log")
       console.error('Error:', error);
     }

utils/CustomPage/Pages/TA/TA.py

@@ -0,0 +1,156 @@
+import json
+from utils.Config import Config
+from utils.CustomPage.CustomObject import CustomObject
+
+import boto3
+from botocore.exceptions import ClientError
+from datetime import datetime
+from collections import defaultdict
+
+class TA(CustomObject):
+    # SHEETS_TO_SKIP = ['Info', 'Appendix']
+    taFindings = {}
+    taError = ''
+    ResourcesToTrack = {}
+    def __init__(self):
+        super().__init__()
+        return
+
+    def build(self):
+        print("... Running CP - TA, it can takes up to 60 seconds")
+        ssBoto = Config.get('ssBoto')
+        ta_client = ssBoto.client('trustedadvisor', region_name='us-east-1')
+        findings = defaultdict(lambda: defaultdict(list))
+
+        try:
+            pillars = ['cost_optimizing', 'security', 'performance', 'fault_tolerance', 'service_limits', 'operational_excellence']
+
+            # First check if user has access to Trusted Advisor
+            try:
+                # Test API access with a simple call
+                ta_client.list_recommendations(pillar='security', maxResults=1)
+            except ClientError as e:
+                error_code = e.response['Error']['Code']
+                if error_code == 'SubscriptionRequiredException':
+                    errMsg = "Error: Your AWS account doesn't have the required Business or Enterprise Support plan for Trusted Advisor access."
+                    self.taError = errMsg
+                    print(errMsg)
+                    return
+                elif error_code in ['AccessDeniedException', 'UnauthorizedOperation']:                    
+                    # errMsg = "Error: You don't have sufficient permissions to access Trusted Advisor. Required IAM permissions: trustedadvisor:List*, trustedadvisor:Get*"
+                    errMsg = e.response['Error']['Message']
+                    self.taError = errMsg
+                    print(errMsg)
+                    return
+                else:
+                    raise e
+
+
+            for pillar in pillars:
+                try:
+                    # Get recommendations for each pillar
+                    recommendations = ta_client.list_recommendations(
+                        pillar=pillar
+                    )['recommendationSummaries']
+
+                    active_recommendations = [
+                        recomm for recomm in recommendations 
+                        if recomm['status'] not in ['resolved', 'dismissed']
+                    ]
+
+                    # Process each recommendation
+                    for recomm in active_recommendations:
+                        # Get detailed recommendation information
+                        detailed_recomm = ta_client.get_recommendation(
+                            recommendationIdentifier=recomm['arn']
+                        )['recommendation']
+
+                        # Create recommendation data structure
+                        recomm_data = {
+                            'name': recomm['name'],
+                            'description': detailed_recomm.get('description', 'N/A'),
+                            'status': recomm['status'],
+                            'source': recomm.get('source', 'N/A'),
+                            'last_updated': recomm.get('lastUpdatedAt', 'N/A'),
+                            'lifecycle_stage': recomm.get('lifecycleStage', 'N/A'),
+                            'error_count': recomm.get('resourcesAggregates', {}).get('errorCount', 0),
+                            'warning_count': recomm.get('resourcesAggregates', {}).get('warningCount', 0),
+                            'ok_count': recomm.get('resourcesAggregates', {}).get('okCount', 0)
+                        }
+
+                        # Add cost optimization specific data
+                        if pillar == 'cost_optimizing':
+                            cost_data = recomm.get('pillarSpecificAggregates', {}).get('costOptimizing', {})
+                            recomm_data.update({
+                                'estimated_savings': cost_data.get('estimatedMonthlySavings', 0),
+                                'estimated_percent_savings': cost_data.get('estimatedPercentMonthlySavings', 0)
+                            })
+
+                        # Group by service
+                        for service in recomm.get('awsServices', ['UNKNOWN']):
+                            findings[pillar][service].append(recomm_data)
+
+                except Exception as e:
+                    errMsg = f"Error processing pillar {pillar}: {str(e)}"
+                    self.taError = errMsg
+                    print(errMsg)
+                    continue
+
+            # Print detailed findings
+            for pillar, services in findings.items():
+                # reset info
+                secTitle = ""
+                secTotal = {"Error": 0, "Warning": 0, "OK": 0}
+                tabDetail = []
+                thead = ["Services", "Findings", "# Error", "# Warning", "# OK", "Last Updated"]
+                if pillar == 'cost_optimizing':
+                    thead.append("Estimated Monthly Savings")
+                    thead.append("Estimated Percent Savings")
+
+                secTitle = pillar.upper()
+                self.taFindings[secTitle] = []
+                rowInfo = []
+                for service, recommendations in services.items():
+                    total_error = sum(r['error_count'] for r in recommendations)
+                    total_warning = sum(r['warning_count'] for r in recommendations)
+                    total_ok = sum(r['ok_count'] for r in recommendations)
+
+                    # print(f"Total Resources: Error({total_error}) Warning({total_warning}) OK({total_ok})")
+                    secTotal['Error'] += total_error    
+                    secTotal['Warning'] += total_warning
+                    secTotal['OK'] += total_ok
+
+                    # Print individual recommendations
+                    for recomm in recommendations:
+                        detail = [service]
+
+                        statClass = 'success'
+                        if recomm['status'] == 'error':
+                            statClass = 'danger'
+                        elif recomm['status'] == 'warning':
+                            statClass = 'warning'
+
+                        statusStr = "<span class='badge badge-{}'>{}</span>".format(statClass, recomm['status'].upper())
+
+                        detail.append(f"{statusStr} {recomm['name']} <i>(Source: {recomm['source']})</i>")
+                        # detail.append(f"{recomm['name']} <i>(Source: {recomm['source']})</i>")
+                        detail.append(f"{recomm['error_count']}")
+                        detail.append(f"{recomm['warning_count']}")
+                        detail.append(f"{recomm['ok_count']}")
+
+                        parsed_datetime = datetime.fromisoformat(f"{recomm['last_updated']}")
+                        formatted_datetime = parsed_datetime.strftime('%Y-%m-%d %H:%M:%S')
+                        detail.append(f"{formatted_datetime} UTC")
+
+                        if pillar == 'cost_optimizing':
+                            detail.append(f"${recomm['estimated_savings']:,.2f}")
+                            detail.append(f"{recomm['estimated_percent_savings']:.1f}%")
+
+                        detail.append(f"{recomm['description']}")
+                        rowInfo.append(detail)
+
+                self.taFindings[secTitle] = [rowInfo, thead, secTotal.copy()]
+                # formatOutput(secTitle, [0,0,0], thead, rowInfo)
+
+        except Exception as e:
+            print(f"Error: {str(e)}")

utils/CustomPage/Pages/TA/TAPageBuilder.py

@@ -0,0 +1,60 @@
+import json
+from utils.CustomPage.CustomPageBuilder import CustomPageBuilder
+from utils.Config import Config
+import constants as _C
+
+class TAPageBuilder(CustomPageBuilder):
+    hasError = False
+    def customPageInit(self):
+        if not self.data.taError == '':
+            self.hasError = True
+
+        return
+
+    def buildContentSummary_customPage(self):
+        if self.hasError:
+            return ["<span>{}</span>".format(self.data.taError)]
+        # info = self.data.taFindings
+
+        pass
+
+    def buildContentDetail_customPage(self):
+        if self.hasError:
+            return
+
+        output = []
+
+        for pillar, pillarInfo in self.data.taFindings.items():
+            rows = pillarInfo[0]
+            head = pillarInfo[1]
+            summ = pillarInfo[2]
+
+            op = self.formatOutput(pillar, head, rows)
+
+            cardTitle = "{} <span class='badge badge-danger'>{} Error</span> <span class='badge badge-warning'>{} Warning</span> <span class='badge badge-success'>{} Ok</span>".format(pillar, summ['Error'], summ['Warning'], summ['OK'])
+
+            card = self.generateCard(pid=self.getHtmlId(pillar), html=op, cardClass='primary', title=cardTitle, titleBadge='', collapse=True, noPadding=False)
+            items = [[card, '']]
+
+            output.append(self.generateRowWithCol(12, items, "data-context='settingTable'"))
+
+        return output
+
+    def formatOutput(self, title, thead, rowInfo):
+        htmlO = []
+        htmlO.append("<table class='table table-bordered table-hover'>")
+
+        fieldSize = len(thead)
+        for headInfo in thead:
+            htmlO.append(f"<th>{headInfo}</th>")
+
+        for row in rowInfo:
+            # print(row)
+            htmlO.append("<tr data-widget='expandable-table' aria-expanded='false'>")
+            for i in range(fieldSize):
+                htmlO.append(f"<td>{row[i]}</td>")
+            htmlO.append("</tr>")
+            htmlO.append("<tr class='expandable-body d-none'><td colspan={}><p style='display: none;'>{}</p></td></tr>".format(fieldSize, row[fieldSize]))
+
+        htmlO.append("</table>")
+        return ''.join(htmlO)