add: tools, secrets (safe files), updated gitignore
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
import json
|
||||
import sys
|
||||
|
||||
def analyze_har(file_path):
|
||||
with open(file_path, 'r', encoding='utf-8') as f:
|
||||
har_data = json.load(f)
|
||||
|
||||
target_uid = "79ac8a4a-353b-4f3f-8ab1-fa937d2d23f8"
|
||||
|
||||
entries = har_data['log']['entries']
|
||||
|
||||
found_responses = []
|
||||
|
||||
for entry in entries:
|
||||
request = entry['request']
|
||||
response = entry['response']
|
||||
|
||||
# Check if the URL contains the UID or the request/response body contains it
|
||||
url = request.get('url', '')
|
||||
|
||||
resp_text = ""
|
||||
if 'content' in response and 'text' in response['content']:
|
||||
resp_text = response['content']['text']
|
||||
|
||||
if target_uid in url or target_uid in resp_text:
|
||||
found_responses.append({
|
||||
'url': url,
|
||||
'method': request['method'],
|
||||
'status': response['status'],
|
||||
'time': entry['startedDateTime'],
|
||||
'response': resp_text
|
||||
})
|
||||
|
||||
# Sort by time
|
||||
found_responses.sort(key=lambda x: x['time'])
|
||||
|
||||
for i, item in enumerate(found_responses):
|
||||
print(f"--- Entry {i+1} ---")
|
||||
print(f"Time: {item['time']}")
|
||||
print(f"Method: {item['method']}")
|
||||
print(f"URL: {item['url']}")
|
||||
print(f"Status: {item['status']}")
|
||||
|
||||
if item['response']:
|
||||
try:
|
||||
data = json.loads(item['response'])
|
||||
# Only print interesting parts to avoid flooding
|
||||
if 'instanceOperation' in data:
|
||||
op = data['instanceOperation']
|
||||
print(f"Operation: {op.get('operation')} Status: {op.get('isSuccessful')} Progress: {op.get('isInProgress')}")
|
||||
if 'stages' in op:
|
||||
for stage in op['stages']:
|
||||
print(f" Stage: {stage.get('stage')} Successs: {stage.get('isSuccessful')} Msg: {stage.get('stageMsg')}")
|
||||
elif 'instance' in data:
|
||||
inst = data['instance']
|
||||
print(f"Instance Status: {inst.get('explainedStatus')}")
|
||||
except:
|
||||
print("Response is not JSON or parsing failed")
|
||||
print("\n")
|
||||
|
||||
if __name__ == "__main__":
|
||||
analyze_har("/home/naeel/terra/har/f12vmbad.har")
|
||||
@@ -0,0 +1,98 @@
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
|
||||
har_path = '/home/naeel/terra/har/faststart.har'
|
||||
|
||||
def analyze_har():
|
||||
with open(har_path, 'r', encoding='utf-8') as f:
|
||||
har_data = json.load(f)
|
||||
|
||||
entries = har_data['log']['entries']
|
||||
|
||||
instance_id = None
|
||||
operation_id = None
|
||||
start_time = None
|
||||
end_time = None
|
||||
error_message = None
|
||||
|
||||
# Sort entries by start time
|
||||
entries.sort(key=lambda x: x['startedDateTime'])
|
||||
|
||||
for entry in entries:
|
||||
req = entry['request']
|
||||
res = entry['response']
|
||||
url = req['url']
|
||||
|
||||
# Look for instance UID in body
|
||||
if req.get('postData', {}).get('text'):
|
||||
if 'instanceUid' in req['postData']['text']:
|
||||
try:
|
||||
body = json.loads(req['postData']['text'])
|
||||
instance_id = body.get('instanceUid')
|
||||
except: pass
|
||||
|
||||
# Look for operation creation or status
|
||||
if '/instanceOperations' in url:
|
||||
if req['method'] == 'POST' and res['status'] == 201:
|
||||
try:
|
||||
body = json.loads(res['content'].get('text', '{}'))
|
||||
operation_id = body.get('uid')
|
||||
start_time = entry['startedDateTime']
|
||||
print(f"Operation started: {operation_id} at {start_time}")
|
||||
except: pass
|
||||
|
||||
# Match the UUID from the URL if not found in POST
|
||||
if not operation_id:
|
||||
parts = url.split('/')
|
||||
for part in parts:
|
||||
if len(part) == 36 and '-' in part: # UUID shape
|
||||
operation_id = part
|
||||
break
|
||||
|
||||
# Track operation progress/status
|
||||
if operation_id and f'/instanceOperations/{operation_id}' in url and req['method'] == 'GET':
|
||||
try:
|
||||
text = res['content'].get('text', '{}')
|
||||
if not text: continue
|
||||
body = json.loads(text)
|
||||
if isinstance(body, list):
|
||||
body = body[0] if body else {}
|
||||
status = body.get('status')
|
||||
print(f"Polling {operation_id}: status={status} at {entry['startedDateTime']}")
|
||||
# print(f"DEBUG: {text[:200]}") # Print first 200 chars
|
||||
if status in ['failed', 'completed', 'success', 'error', 'failed-manual-fix']:
|
||||
if not end_time or entry['startedDateTime'] > end_time:
|
||||
end_time = entry['startedDateTime']
|
||||
error_message = body.get('error') or body.get('message') or body.get('statusMessage')
|
||||
if not error_message and body.get('steps'):
|
||||
# Check steps for errors
|
||||
for step in body.get('steps'):
|
||||
if step.get('status') == 'failed':
|
||||
error_message = step.get('error') or step.get('statusMessage')
|
||||
print(f"Operation reached terminal status {status} at {end_time}")
|
||||
print(f"Full response: {text}")
|
||||
except Exception as e:
|
||||
# print(f"Error parsing: {e}")
|
||||
pass
|
||||
|
||||
# Check for Sun/Date anomalies in headers
|
||||
for header in res['headers']:
|
||||
if header['name'].lower() == 'date':
|
||||
if 'Sun' in header['value']:
|
||||
print(f"Strange Date header found: {header['value']} in {url}")
|
||||
|
||||
print(f"\nSummary:")
|
||||
print(f"Instance ID: {instance_id}")
|
||||
print(f"Operation ID: {operation_id}")
|
||||
if start_time and end_time:
|
||||
t1 = datetime.fromisoformat(start_time.replace('Z', '+00:00'))
|
||||
t2 = datetime.fromisoformat(end_time.replace('Z', '+00:00'))
|
||||
duration = (t2 - t1).total_seconds()
|
||||
print(f"Duration: {duration} seconds")
|
||||
else:
|
||||
print("Could not determine duration")
|
||||
print(f"Error: {error_message}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
analyze_har()
|
||||
@@ -0,0 +1,47 @@
|
||||
import json
|
||||
import os
|
||||
|
||||
def extract_params_from_har(har_path):
|
||||
print(f"\n--- Analyzing {os.path.basename(har_path)} ---")
|
||||
if not os.path.exists(har_path):
|
||||
print(f"File not found: {har_path}")
|
||||
return
|
||||
|
||||
with open(har_path, 'r', encoding='utf-8') as f:
|
||||
har_data = json.load(f)
|
||||
|
||||
entries = har_data['log']['entries']
|
||||
|
||||
# Track operations to group parameters
|
||||
operations = {} # UID -> list of params
|
||||
|
||||
for entry in entries:
|
||||
req = entry['request']
|
||||
# Looking for POST /api/v1/index.cfm/instanceOperationCfsParams
|
||||
if '/instanceOperationCfsParams' in req['url'] and req['method'] == 'POST':
|
||||
post_data = req.get('postData', {})
|
||||
text = post_data.get('text')
|
||||
if text:
|
||||
try:
|
||||
p = json.loads(text)
|
||||
op_uid = p.get('instanceOperationUid')
|
||||
if op_uid not in operations:
|
||||
operations[op_uid] = []
|
||||
operations[op_uid].append(p)
|
||||
except:
|
||||
pass
|
||||
|
||||
for op_uid, params in operations.items():
|
||||
print(f"\nOperation Uid: {op_uid}")
|
||||
for p in params:
|
||||
param_id = p.get('svcOperationCfsParamId')
|
||||
param_val = p.get('paramValue')
|
||||
print(f" ID: {param_id:4} | Value: {repr(param_val)}")
|
||||
|
||||
har_files = [
|
||||
'/home/naeel/terra/har/f12vmbad.har',
|
||||
'/home/naeel/terra/har/f12vmbad1.har'
|
||||
]
|
||||
|
||||
for h in har_files:
|
||||
extract_params_from_har(h)
|
||||
@@ -0,0 +1,22 @@
|
||||
import json
|
||||
|
||||
har_path = '/home/naeel/terra/har/faststart.har'
|
||||
op_id = 'c2e44fae-2a05-4955-867c-763da66b578e'
|
||||
|
||||
with open(har_path, 'r', encoding='utf-8') as f:
|
||||
har_data = json.load(f)
|
||||
|
||||
entries = har_data['log']['entries']
|
||||
|
||||
for entry in entries:
|
||||
url = entry['request']['url']
|
||||
if op_id in url and entry['request']['method'] == 'GET':
|
||||
res = entry['response']
|
||||
text = res['content'].get('text')
|
||||
if text:
|
||||
try:
|
||||
data = json.loads(text)
|
||||
print(f"Time: {entry['startedDateTime']}")
|
||||
print(json.dumps(data, indent=2))
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
@@ -0,0 +1,33 @@
|
||||
import json
|
||||
import os
|
||||
|
||||
def check_creation_status(har_path):
|
||||
print(f"--- Checking status in {os.path.basename(har_path)} ---")
|
||||
with open(har_path, 'r', encoding='utf-8') as f:
|
||||
har_data = json.load(f)
|
||||
|
||||
entries = har_data['log']['entries']
|
||||
|
||||
for entry in entries:
|
||||
req = entry['request']
|
||||
resp = entry['response']
|
||||
|
||||
# Check instanceOperations POST
|
||||
if '/instanceOperations' in req['url'] and req['method'] == 'POST' and not '/instanceOperationCfsParams' in req['url']:
|
||||
print(f"Found Operation POST: {req['url']}")
|
||||
post_data = req.get('postData', {})
|
||||
print(f"Request body: {post_data.get('text')}")
|
||||
print(f"Response status: {resp['status']}")
|
||||
resp_content = resp.get('content', {})
|
||||
print(f"Response body: {resp_content.get('text')}")
|
||||
print("-" * 20)
|
||||
|
||||
har_files = [
|
||||
'/home/naeel/terra/har/f12vmbad.har',
|
||||
'/home/naeel/terra/har/f12vmbad1.har',
|
||||
'/home/naeel/terra/har/faststart.har'
|
||||
]
|
||||
|
||||
for h in har_files:
|
||||
if os.path.exists(h):
|
||||
check_creation_status(h)
|
||||
@@ -0,0 +1,94 @@
|
||||
import json
|
||||
import re
|
||||
|
||||
def get_ids_and_failures(file_path):
|
||||
with open(file_path, 'r', encoding='utf-8') as f:
|
||||
har_data = json.load(f)
|
||||
|
||||
entries = har_data['log']['entries']
|
||||
|
||||
instance_ids = set()
|
||||
operation_ids = set()
|
||||
dates = set()
|
||||
|
||||
# Try to find UUIDs that look like instance or operation IDs
|
||||
# Usually in URLs or body
|
||||
for entry in entries:
|
||||
req = entry['request']
|
||||
resp = entry['response']
|
||||
|
||||
# Check Date header
|
||||
for header in resp.get('headers', []):
|
||||
if header['name'].lower() == 'date':
|
||||
dates.add(header['value'])
|
||||
|
||||
url = req.get('url', '')
|
||||
# Pattern for UUID
|
||||
uuids = re.findall(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', url)
|
||||
for u in uuids:
|
||||
instance_ids.add(u)
|
||||
|
||||
resp_text = ""
|
||||
if 'content' in resp and 'text' in resp['content']:
|
||||
resp_text = resp['content']['text']
|
||||
# Look for instanceOperation or similar
|
||||
if 'instanceOperation' in resp_text:
|
||||
op_uuids = re.findall(r'"id":"([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})"', resp_text)
|
||||
for u in op_uuids:
|
||||
operation_ids.add(u)
|
||||
|
||||
# Look for instance id
|
||||
inst_uuids = re.findall(r'"instanceId":"([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})"', resp_text)
|
||||
for u in inst_uuids:
|
||||
instance_ids.add(u)
|
||||
|
||||
print(f"File: {file_path}")
|
||||
print(f"Possible Instance IDs: {instance_ids}")
|
||||
print(f"Possible Operation IDs: {operation_ids}")
|
||||
|
||||
# Analyze flow
|
||||
print("\nFlow analysis (Status 400+ or instanceOperation):")
|
||||
for entry in entries:
|
||||
req = entry['request']
|
||||
resp = entry['response']
|
||||
status = resp['status']
|
||||
url = req['url']
|
||||
time = entry['startedDateTime']
|
||||
|
||||
if status >= 400:
|
||||
print(f"[{time}] ERROR {status} {req['method']} {url}")
|
||||
if 'content' in resp and 'text' in resp['content']:
|
||||
print(f" Response: {resp['content']['text'][:500]}")
|
||||
|
||||
if 'instanceOperation' in url or ('content' in resp and 'text' in resp['content'] and 'instanceOperation' in resp['content']['text']):
|
||||
print(f"[{time}] OP_POLL {status} {url}")
|
||||
if 'content' in resp and 'text' in resp['content']:
|
||||
try:
|
||||
data = json.loads(resp['content']['text'])
|
||||
if isinstance(data, dict):
|
||||
op = data.get('instanceOperation', data)
|
||||
if isinstance(op, dict):
|
||||
success = op.get('isSuccessful')
|
||||
if success is False:
|
||||
# Print summary of stages
|
||||
stages = op.get('stages', [])
|
||||
print(f" FAILED OP STAGES:")
|
||||
for s in stages:
|
||||
print(f" - {s.get('displayName')} Status: {s.get('isSuccessful')} Msg: {s.get('message')}")
|
||||
print(f" FULL ERROR LOG: {op.get('errorLog')}")
|
||||
else:
|
||||
print(f" Op: {op.get('operation')} Success: {success} Progress: {op.get('isInProgress')} Msg: {op.get('message')}")
|
||||
except:
|
||||
pass
|
||||
|
||||
print("\n--- Date headers summary ---")
|
||||
sun_mon = [d for d in dates if "Sun" in d or "Mon" in d]
|
||||
if sun_mon:
|
||||
print(f"Found Sun/Mon dates: {sun_mon[:5]}")
|
||||
else:
|
||||
print("No Sun/Mon dates found in headers.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
if len(sys.argv) > 1:
|
||||
get_ids_and_failures(sys.argv[1])
|
||||
@@ -0,0 +1,42 @@
|
||||
import json
|
||||
import os
|
||||
|
||||
def extract_params_from_har(har_path):
|
||||
print(f"--- Analyzing {os.path.basename(har_path)} ---")
|
||||
with open(har_path, 'r', encoding='utf-8') as f:
|
||||
har_data = json.load(f)
|
||||
|
||||
entries = har_data['log']['entries']
|
||||
|
||||
# Store parameters group by operation (if possible)
|
||||
# Looking for POST /api/v1/index.cfm/instanceOperationCfsParams
|
||||
|
||||
results = []
|
||||
for entry in entries:
|
||||
req = entry['request']
|
||||
if '/instanceOperationCfsParams' in req['url'] and req['method'] == 'POST':
|
||||
post_data = req.get('postData', {})
|
||||
text = post_data.get('text')
|
||||
if text:
|
||||
try:
|
||||
params = json.loads(text)
|
||||
results.append(params)
|
||||
except:
|
||||
pass
|
||||
|
||||
for p in results:
|
||||
# Expected keys: instanceOperationUid, svcOperationCfsParamId, paramValue
|
||||
op_uid = p.get('instanceOperationUid')
|
||||
param_id = p.get('svcOperationCfsParamId')
|
||||
param_val = p.get('paramValue')
|
||||
print(f"Op: {op_uid} | ID: {param_id} | Value: {param_val} (Type: {type(param_val).__name__})")
|
||||
|
||||
har_files = [
|
||||
'/home/naeel/terra/har/vm.har'
|
||||
]
|
||||
|
||||
for h in har_files:
|
||||
if os.path.exists(h):
|
||||
extract_params_from_har(h)
|
||||
else:
|
||||
print(f"File not found: {h}")
|
||||
@@ -0,0 +1,51 @@
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
|
||||
def find_param_defs(har_path):
|
||||
param_map = {}
|
||||
with open(har_path, 'r', encoding='utf-8') as f:
|
||||
har_data = json.load(f)
|
||||
|
||||
for entry in har_data['log']['entries']:
|
||||
resp = entry['response']
|
||||
if 'content' in resp and 'text' in resp['content']:
|
||||
text = resp['content']['text']
|
||||
# Search for pattern "svcOperationCfsParamId":X,"svcOperationCfsParam":"NAME"
|
||||
# Or "svcOperationCfsParamId":X ... "svcOperationCfsParam":"NAME"
|
||||
# Matches in JSON structure
|
||||
try:
|
||||
data = json.loads(text)
|
||||
def walk(obj):
|
||||
if isinstance(obj, dict):
|
||||
if 'svcOperationCfsParamId' in obj and 'svcOperationCfsParam' in obj:
|
||||
param_map[obj['svcOperationCfsParamId']] = {
|
||||
'name': obj['svcOperationCfsParam'],
|
||||
'label': obj.get('label'),
|
||||
'dataType': obj.get('dataType'),
|
||||
'isRequired': obj.get('isRequired')
|
||||
}
|
||||
for v in obj.values():
|
||||
walk(v)
|
||||
elif isinstance(obj, list):
|
||||
for item in obj:
|
||||
walk(item)
|
||||
walk(data)
|
||||
except:
|
||||
pass
|
||||
return param_map
|
||||
|
||||
har_files = [
|
||||
'/home/naeel/terra/har/f12vmbad.har',
|
||||
'/home/naeel/terra/har/f12vmbad1.har',
|
||||
'/home/naeel/terra/har/faststart.har'
|
||||
]
|
||||
|
||||
all_params = {}
|
||||
for h in har_files:
|
||||
if os.path.exists(h):
|
||||
all_params.update(find_param_defs(h))
|
||||
|
||||
# Print mapping
|
||||
for pid in sorted(all_params.keys()):
|
||||
print(f"{pid}: {all_params[pid]}")
|
||||
Reference in New Issue
Block a user