"""
CodeTrack v. 0.99.3 - A tool to track software defects, bug reports, and
change requests.
Python port of the original PHP application. Uses Flask as the web
framework and stores data in simple XML text files (no database required).
See http://www.openbugs.org/ for the original CodeTrack.
This software is released under the GPL and is copyrighted by Kenneth White,
2001-2006. Complete license is LICENSE.txt in the docs directory.
local rename is esltrack
"""
import os
import re
import time
import base64
import hashlib
import pickle
import shutil
import random
import string
import urllib.parse
from datetime import datetime, timedelta
from flask import (
Flask, request, redirect, make_response, Response, render_template_string
)
import config as CT
app = Flask(__name__)
# ---------------------------------------------------------------------------
# Global state (mirrors PHP globals)
# ---------------------------------------------------------------------------
xml_array_g = []
current_session_g = {}
debug_g = False
# ===========================================================================
# Utility helpers
# ===========================================================================
def no_cache():
"""Emit no-cache headers (PHP no_cache())."""
# Handled via response headers in Flask; see helper below.
pass
def _no_cache_response(resp):
resp.headers["Expires"] = "Mon, 26 Jul 1997 05:00:00 GMT"
resp.headers["Last-Modified"] = datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S GMT")
resp.headers["Cache-Control"] = "no-cache, must-revalidate"
resp.headers["Pragma"] = "no-cache"
return resp
def calc_session_id(user_full_name):
"""Create a one-day session ID (MD5 of date + name + private key)."""
if not CT.CT_PRIVATE_KEY:
raise RuntimeError("Fatal: Internal - no authentication key!")
if not user_full_name:
return False
todays_sid = datetime.now().strftime("%d%b%Y") + user_full_name + CT.CT_PRIVATE_KEY
return hashlib.md5(todays_sid.encode("utf-8")).hexdigest()
def build_new_session_cookie(user_data, selected_project):
if not user_data or not selected_project:
raise RuntimeError("Fatal-Internal: No User Data or Selected Project!")
session_auth = calc_session_id(user_data["Full_Name"])
session = {
"user_full_name": user_data["Full_Name"],
"id": session_auth,
"project": selected_project,
"username": user_data["Username"],
"role": user_data["Role"],
}
serialized = urllib.parse.quote(base64.b64encode(pickle.dumps(session)).decode("utf-8"))
return serialized
def build_updated_session_cookie(project_name, current_session):
session = {
"user_full_name": current_session["user_full_name"],
"id": current_session["id"],
"project": project_name,
"username": current_session["username"],
"role": current_session["role"],
}
serialized = urllib.parse.quote(base64.b64encode(pickle.dumps(session)).decode("utf-8"))
return serialized
def decode_session_cookie(cookie):
try:
return pickle.loads(base64.b64decode(urllib.parse.unquote(cookie)))
except Exception:
return None
def set_session_cookie_load_page(serialized_session, page_to_load):
"""Set session cookie and redirect (PHP used JS; we use a redirect)."""
resp = make_response(redirect(f"/esltrack.php?page={page_to_load}"))
resp.set_cookie("esltrack_session", serialized_session, path="/")
return resp
def calc_next_node_id(node_list):
max_id = 0
for node in node_list:
if int(node.get("ID", 0)) > max_id:
max_id = int(node["ID"])
return max_id + 1
def get_array_index_for_id(data_array, id_):
row_cnt = len(data_array)
if row_cnt == 1:
return 0
for index in range(row_cnt):
if str(data_array[index].get("ID", "")) == str(id_):
return index
raise RuntimeError(f"Fatal: No issue matching ID# {id_} found!")
def get_next_bug_id(bug_array, current_row_id):
n = current_row_id
next_id = None
while True:
n += 1
if n >= len(bug_array):
break
if ("Delete_Bug" not in bug_array[n]
and bug_array[n].get("Project") == current_session_g.get("project")
and "Closed" not in bug_array[n].get("Status", "")):
next_id = bug_array[n]["ID"]
break
if next_id is None:
next_id = bug_array[current_row_id]["ID"]
return next_id
def get_prev_bug_id(bug_array, current_row_id):
p = current_row_id
prev_id = None
while True:
p -= 1
if p < 0:
break
if ("Delete_Bug" not in bug_array[p]
and bug_array[p].get("Project") == current_session_g.get("project")
and "Closed" not in bug_array[p].get("Status", "")):
prev_id = bug_array[p]["ID"]
break
if prev_id is None:
prev_id = bug_array[current_row_id]["ID"]
return prev_id
def authorized_user(project_id, user_id, permission_table):
for permission in permission_table:
if (permission.get("Project_ID") == str(project_id)
and permission.get("User_ID") == str(user_id)):
return True
return False
def admin_check():
if current_session_g.get("role") != "Admin":
raise RuntimeError("
You are not an administrator.")
def get_stats(time_in_seconds):
days = time_in_seconds / 86400.0
if days:
return "%3.1f days" % days
return "(n/a)"
def parse_human_date(date_string):
try:
dt = datetime.strptime(date_string, CT.CT_DATE_FORMAT)
except Exception:
return date_string
today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
if dt > today:
return "Today, " + dt.strftime("%I:%M %p")
if dt > today - timedelta(days=1):
return "Yesterday, " + dt.strftime("%I:%M %p")
return date_string
def reverse_htmlspecialchars(html_encoded_string):
"""Undo htmlspecialchars: " -> \" etc."""
trans = {
""": '"', "'": "'", "<": "<", ">": ">", "&": "&",
}
for k, v in trans.items():
html_encoded_string = html_encoded_string.replace(k, v)
return html_encoded_string
def encrypt_password(plaintext_password):
"""Produce an alphanumeric-only MD5-based hash (PHP crypt() equivalent)."""
sanity_check = 0
while True:
salt = "".join(chr(random.randint(48, 122)) for _ in range(2))
# Emulate PHP crypt() with a 2-char salt -> use MD5-based crypt
import crypt as _crypt
encrypted = _crypt.crypt(plaintext_password, salt)
if re.search(r"[^0-9a-z]", encrypted, re.I) is None:
return encrypted
sanity_check += 1
if sanity_check > 250:
raise RuntimeError("Fatal: Could not construct valid encrypted password!")
def valid_password(login_data, user_table):
"""Return (match, index)."""
user_cnt = len(user_table)
match = False
index = -1
for i in range(user_cnt):
if login_data.get("username") == user_table[i].get("Username"):
plaintext = login_data.get("password", "")
hashed = user_table[i].get("Password", "")
salt = hashed[:2]
import crypt as _crypt
match = (_crypt.crypt(plaintext, salt) == hashed)
index = i
break
return match, index
# ===========================================================================
# File locking (brute-force lock files)
# ===========================================================================
def brute_force_lock(filename):
brute_force_lock_check(filename)
lockfile = filename + ".lck"
try:
open(lockfile, "a").close()
except OSError:
raise RuntimeError(
f"
Fatal: Could not create file lock {lockfile} ! "
" Make sure the web server has write permissions in this directory")
def brute_force_lock_check(filename):
lockfile = filename + ".lck"
timeout = 0
while os.path.exists(lockfile):
time.sleep(1)
if timeout > CT.CT_LOCKFILE_TIMEOUT:
brute_force_lock_release(filename)
raise RuntimeError(
f"
Fatal: Timed out waiting on file lock {lockfile} !"
" Lock has been cleared. Please report this message to support.")
timeout += 1
def brute_force_lock_release(filename):
lockfile = filename + ".lck"
try:
os.unlink(lockfile)
except OSError:
raise RuntimeError(
f"
Fatal: Could not release file lock {lockfile} ! "
"Please report this message to support.")
# ===========================================================================
# XML parsing / writing
# ===========================================================================
def parse_xml_file(xml_filename, drop_history=False, sort_element="", sort_order=""):
"""Non-validating single-pass XML parser. Populates global xml_array_g."""
global xml_array_g
xml_array_g = []
try:
with open(xml_filename, "r", encoding="utf-8", errors="replace") as fp:
xml_document = fp.read()
except OSError:
raise RuntimeError(
f"
Fatal: Error opening '{xml_filename}'. "
"Make sure it's readable by the web server owner.")
# Pull out (data)
matches = re.findall(r"<([\w]+)[^>]*>([^<]*)", xml_document)
node_cnt = 0
node_id = 0
index = 0
for tag, content in matches:
if tag == "ID":
node_id = int(content) if content.isdigit() else 0
if drop_history:
index = node_id
if node_id < len(xml_array_g):
xml_array_g[node_id] = {}
else:
while len(xml_array_g) <= node_id:
xml_array_g.append({})
else:
index = node_cnt
xml_array_g.append({})
node_cnt += 1
if index < len(xml_array_g):
xml_array_g[index][tag] = content
# Remove empty dicts left over from drop_history gaps
xml_array_g = [x for x in xml_array_g if x]
if sort_element:
if len(xml_array_g) == 1:
pass
elif sort_order == CT.CT_ASCENDING:
xml_array_g.sort(key=lambda x: x.get(sort_element, "").lower())
else:
xml_array_g.sort(key=lambda x: x.get(sort_element, "").lower(), reverse=True)
def parse_dtd_file(filename):
try:
with open(filename, "r", encoding="utf-8", errors="replace") as fh:
dtd = fh.read()
except OSError:
raise RuntimeError(
f"
Fatal: Couldn't open '{filename}' ! "
"Check that it's readable by the web server owner.")
matches = re.findall(r"", dtd)
return matches
def append_xml_file_node(filename, node, closing_root_tag):
brute_force_lock(filename)
try:
with open(filename, "rb+") as fh:
fh.seek(0, os.SEEK_END)
fsize = fh.tell()
fh.seek(-CT.CT_LAST_LINE_SIZE, os.SEEK_END)
last_line = fh.read(CT.CT_LAST_LINE_SIZE).decode("utf-8", errors="replace")
except OSError:
raise RuntimeError(
f"
Fatal: Couldn't append to '{filename}' ! "
"Check that it's writable by the web server owner.")
pos = last_line.rfind(closing_root_tag)
offset = CT.CT_LAST_LINE_SIZE - pos if pos != -1 else 0
append_location = fsize - offset
node += "\n" + closing_root_tag
# Translate bare \n to \r\n for friendly transfers
node = re.sub(r"([^\r])\n", r"\1\r\n", node)
try:
with open(filename, "rb+") as fh:
fh.seek(append_location)
fh.write(node.encode("utf-8"))
except OSError:
raise RuntimeError(f"Fatal: Unable to add XML node to '{filename}' !")
brute_force_lock_release(filename)
def rewrite_xml_file(filename, parsed_tree, major_element_tag):
root_element_tag = major_element_tag + "s"
brute_force_lock(filename)
opening_content = (
'\r\n'
f"\r\n"
"\r\n"
f'\r\n'
f"<{root_element_tag}>\r\n"
)
try:
with open(filename, "wb") as fh:
fh.write(opening_content.encode("utf-8"))
for node in parsed_tree:
fh.write(f"\t<{major_element_tag}>\r\n".encode("utf-8"))
for element, contents in node.items():
fh.write(f"\t\t<{element}>{contents}\r\n".encode("utf-8"))
fh.write(f"\t\r\n".encode("utf-8"))
fh.write(f"".encode("utf-8"))
except OSError:
raise RuntimeError(f"Fatal: Unable to write XML data to '{filename}' !")
brute_force_lock_release(filename)
def update_xml_file_node(filename, parsed_tree, major_element_tag, update_node):
node_cnt = len(parsed_tree)
target_id = update_node["ID"]
node_updated = False
for node_index in range(node_cnt):
if parsed_tree[node_index].get("ID") == target_id:
for element, contents in update_node.items():
parsed_tree[node_index][element] = contents
node_updated = True
if not node_updated:
raise RuntimeError(f"
Fatal: XML Node ID {target_id} not found! ")
rewrite_xml_file(filename, parsed_tree, major_element_tag)
# ===========================================================================
# Data scrubbing
# ===========================================================================
def scrub_and_tag_form_data(data_array):
for element in list(data_array.keys()):
untrusted_content = data_array[element]
if untrusted_content == "":
del data_array[element]
else:
untrusted_content = str(untrusted_content)[:CT.CT_MAX_DESCR_SIZE].strip()
content = re.sub(
r'[^ 0-9a-z_@#$%();:?+*!=&,"\'\n\t.-]+', "", untrusted_content, flags=re.I)
clean_data = f"<{element}>" + _htmlspecialchars(content.strip()) + f""
data_array[element] = clean_data
return data_array
def _htmlspecialchars(s):
return (s.replace("&", "&").replace('"', """)
.replace("'", "'").replace("<", "<").replace(">", ">"))
def process_file_attachment(tmp_filename, requested_filename):
filename = re.sub(r"( +)", "_", requested_filename.strip())
filename = re.sub(r"([^.0-9a-z_-]+)", "", filename, flags=re.I)
filename = re.sub(r"(\.+)", ".", filename)
if len(filename) > CT.CT_MAX_ATTACHMENT_FILENAME_SIZE:
raise RuntimeError(
"\n
The filename of your attachment is too long. "
f"The name cannot exceed {CT.CT_MAX_ATTACHMENT_FILENAME_SIZE} characters.
"
"* Please contact the esltrack admin if this is incorrect. * "
' ( check "CT_MAX_ATTACHMENT_FILENAME_SIZE" setting )')
legal_attachment_types = [x.strip() for x in CT.CT_LEGAL_ATTACHMENTS.split(",")]
legal = False
for legal_type in legal_attachment_types:
if re.search(r"\.%s$" % re.escape(legal_type), filename, re.I):
legal = True
break
if not legal:
raise RuntimeError(
"\n
Your attachment is not a permissible file.
"
f"Legal file types: {CT.CT_LEGAL_ATTACHMENTS}
"
' * Please contact the esltrack admin if this is incorrect (check "CT_LEGAL_ATTACHMENTS" setting) *')
os.makedirs(CT.CT_ATTACHMENTS_DIRECTORY, exist_ok=True)
dest = os.path.join(CT.CT_ATTACHMENTS_DIRECTORY, filename)
try:
shutil.move(tmp_filename, dest)
os.chmod(dest, 0o600)
return filename
except OSError:
raise RuntimeError(
f" Fatal: File attachment transfer failed for TMP: {tmp_filename} "
f"FILE: {filename}
"""
def draw_login_page(project_table, failed="", session_expired="", no_cookies=""):
project_cnt = len(project_table)
options = ""
for i in range(project_cnt):
title = project_table[i]["Title"]
selected = " selected='selected' " if CT.CT_DEFAULT_PROJECT == title else ""
options += f"\t\t\n"
out = f"""
"""
if failed:
out += "
Your login failed.
Please enter a correct username and password.
\n"
if session_expired:
out += "
Your session has expired.
Please log in again.
\n"
if no_cookies:
out += ("
Login unsuccessful.
"
"esltrack requires session cookies. Please ensure they are not being blocked.
\n")
out += """
If you can see this, there are permission problems on the style directory.
If you can see this, there are permission problems on the images directory.
If you can see this, there are permission problems on the javascript directory.