142 lines
5.1 KiB
Python
142 lines
5.1 KiB
Python
import yaml
|
||
import re
|
||
from datetime import datetime
|
||
from flask import Flask, request, jsonify
|
||
|
||
app = Flask(__name__)
|
||
|
||
# Load GPO data
|
||
with open('{{ gpo_yaml_path }}', 'r') as file:
|
||
data = yaml.safe_load(file)
|
||
|
||
def strip_dn(dn: str, base_suffix: str = ",OU=Manufacturing,OU=Tesla Systems,DC=teslamotors,DC=com") -> str:
|
||
suffix = base_suffix.strip()
|
||
dn_clean = dn.strip()
|
||
lowered_dn = dn_clean.lower()
|
||
lowered_suffix = suffix.lower()
|
||
idx = lowered_dn.rfind(lowered_suffix)
|
||
if idx != -1:
|
||
dn_clean = dn_clean[:idx].rstrip(',')
|
||
|
||
# Pull all OU=… values (leaf → root order)
|
||
ou_values: List[str] = re.findall(r'OU=([^,]+)', dn_clean, flags=re.IGNORECASE)
|
||
|
||
if not ou_values:
|
||
return ''
|
||
ou_values = [v.strip() for v in reversed(ou_values)]
|
||
return '\\'.join(ou_values)
|
||
|
||
@app.route('/gpo', methods=['GET'])
|
||
def get_groups():
|
||
group_name = request.args.get('group')
|
||
print("Group specified:", group_name)
|
||
if not group_name:
|
||
return jsonify({"error": "Group parameter is required"}), 400
|
||
|
||
gpos = []
|
||
for entry in data:
|
||
configured_gpos = entry.get('configured_gpos', {})
|
||
for gpo, details in configured_gpos.items():
|
||
if details is None:
|
||
continue
|
||
for detail in details:
|
||
# Check if 'ad_group' exists and is not None before accessing it
|
||
ad_group = detail.get('ad_group')
|
||
# this allows for partial search
|
||
if isinstance(ad_group, str) and re.search(re.escape(group_name.lower()), ad_group.lower()):
|
||
# i don't like BUILTIN\ so I'm getting rid of it
|
||
local_group = detail.get('local_group')
|
||
if isinstance(local_group, str) and local_group.startswith("BUILTIN\\"):
|
||
local_group = local_group.split("\\", 1)[1]
|
||
gpos.append({
|
||
"gpo": gpo,
|
||
"local_group": local_group,
|
||
"ad_group": detail['ad_group']
|
||
})
|
||
return jsonify(gpos)
|
||
|
||
@app.route("/gpo_links", methods=["GET"])
|
||
def gpo_links():
|
||
try:
|
||
gpo_link_count = data[0]['gpo_links']
|
||
except:
|
||
gpo_link_count = "whoops"
|
||
return jsonify({"gpo_links": gpo_link_count})
|
||
|
||
@app.route("/timestamp", methods=["GET"])
|
||
def timestamp():
|
||
try:
|
||
yaml_timestamp = data[0]['timestamp']
|
||
except:
|
||
yaml_timestamp = "whoops"
|
||
return jsonify({"timestamp": yaml_timestamp})
|
||
|
||
@app.route("/index_duration", methods=["GET"])
|
||
def index_duration():
|
||
fmt="%m/%d/%Y %I:%M:%S %p"
|
||
try:
|
||
start_timestamp = data[0]['init_timestamp']
|
||
end_timestamp = data[0]['timestamp']
|
||
duration = (datetime.strptime(end_timestamp, fmt) - datetime.strptime(start_timestamp, fmt)).total_seconds() / 60
|
||
return jsonify({"duration": duration, "start_timestamp": start_timestamp, "end_timestamp": end_timestamp})
|
||
except:
|
||
return jsonify({'duration': "whoops"})
|
||
|
||
@app.route("/linked_ous", methods=["GET"])
|
||
def linked_ous():
|
||
gpo_name = request.args.get("gpo")
|
||
if not gpo_name:
|
||
return jsonify({"error": "Missing 'gpo' query parameter"}), 400
|
||
linked = []
|
||
# The YAML is a list of dictionaries – iterate over them
|
||
|
||
for top in data or []:
|
||
cfg = top.get("configured_gpos", {})
|
||
if not isinstance(cfg, dict):
|
||
continue
|
||
|
||
# Grab the value for the requested GPO – it can be a dict *or* a list
|
||
gpo_entry = cfg.get(gpo_name)
|
||
if gpo_entry is None:
|
||
continue
|
||
|
||
def _add_ous(ous):
|
||
if not isinstance(ous, list):
|
||
return
|
||
for dn in ous:
|
||
if isinstance(dn, str):
|
||
# 1. strip out OU/ DC components
|
||
# 2. keep the *leaf* and its *parent* (e.g. NA\SJC18)
|
||
stripped = strip_dn(dn)
|
||
if stripped:
|
||
linked.append(stripped)
|
||
|
||
# Case 1: direct dict → look for Linked_OUs key
|
||
if isinstance(gpo_entry, dict):
|
||
if "Linked_OUs" in gpo_entry:
|
||
_add_ous(gpo_entry["Linked_OUs"])
|
||
continue
|
||
|
||
# Case 2: list of dicts → find the dict that has the Linked_OUs key
|
||
if isinstance(gpo_entry, list):
|
||
for sub in gpo_entry:
|
||
if isinstance(sub, dict) and "Linked_OUs" in sub:
|
||
_add_ous(sub["Linked_OUs"])
|
||
|
||
return jsonify(linked)
|
||
|
||
# test route
|
||
@app.route('/test', methods=['GET'])
|
||
def test():
|
||
try:
|
||
root_ou = data[0]['root_ou']
|
||
except KeyError:
|
||
# Handle the case where 'root_ou' does not exist in the dictionary
|
||
return jsonify({"error": "KeyError: 'root_ou' not found in provided data"}), 400
|
||
except TypeError:
|
||
# Handle cases where data might be of a different type than expected
|
||
return jsonify({"error": "TypeError: Data is not a dictionary"}), 400
|
||
return jsonify({"result": "test", "OU_Root": root_ou})
|
||
|
||
if __name__ == '__main__':
|
||
app.run(debug=True, host='{{ api_service_bind_ip }}', port={{ api_service_port }}) |