import json

import re

from typing import Callable, Optional

from requests import Response

from socketsecurity.core import log

from socketsecurity.core.classes import Comment, Issue

from socketsecurity.core.messages import Messages

class Comments:

VIEW_REPORT_PATTERN = re.compile(r"\[View full report\]\(([^)\s]+)\)")

@staticmethod

def comment_author_name(comment: Comment) -> str:

"""Best-effort display name for a comment author, across providers."""

user = getattr(comment, "user", None) or getattr(comment, "author", None) or {}

return user.get("login") or user.get("username") or "an unknown user"

@staticmethod

def process_response(response: Response) -> dict:

output = {}

try:

output = response.json()

except Exception as error:

log.debug("Unable to parse comment response json, trying as text")

log.debug(error)

try:

output = json.loads(response.text)

except Exception as error:

log.error("Unable to process comment data, unable to get previous comment data")

log.error(error)

return output

@staticmethod

def remove_alerts(comments: dict, new_alerts: list) -> list:

alerts = []

if "ignore" not in comments:

return new_alerts

ignore_all, ignore_commands = Comments.get_ignore_options(comments)

for alert in new_alerts:

alert: Issue

if ignore_all:

break

else:

if any(

Comments.is_ignore(alert.pkg_name, alert.pkg_version, name, version, alert.pkg_type)

for name, version in ignore_commands

):

log.info(f"Alerts for {alert.pkg_name}@{alert.pkg_version} ignored")

else:

log.info(f"Adding alert {alert.type} for {alert.pkg_name}@{alert.pkg_version}")

alerts.append(alert)

return alerts

@staticmethod

def get_ignore_options(comments: dict) -> [bool, list]:

ignore_commands = []

ignore_all = False

for comment in comments["ignore"]:

comment: Comment

first_line = comment.body_list[0]

if not ignore_all and "socketsecurity ignore" in first_line.lower():

try:

first_line = first_line.lstrip("@")

# Case-insensitive split: find "SocketSecurity " regardless of casing

lower_line = first_line.lower()

split_idx = lower_line.index("socketsecurity ") + len("socketsecurity ")

command = first_line[split_idx:].strip()

if command == "ignore-all":

ignore_all = True

else:

command = command.lstrip("ignore").strip()

name, separator, version = command.rpartition("@")

if not separator or not name or not version:

raise ValueError("Expected package@version")

data = (name.strip(), version.strip())

ignore_commands.append(data)

except Exception as error:

log.error(f"Unable to process ignore command for {comment}")

log.error(error)

return ignore_all, ignore_commands

@staticmethod

def is_ignore(

pkg_name: str, pkg_version: str, name: str, version: str,

pkg_type: str = ""

) -> bool:

"""Match an alert's package against one parsed ignore command.

Generated commands are ecosystem-qualified (``npm/lodash@4.17.21``) but

replies typed by hand, and commands written by older CLI versions, use the

bare package name, so both have to match.

Callers that parse the package out of a ``start-socket-alert`` marker have no

pkg_type to compare against and instead strip the ecosystem off the command.

An npm scope looks the same as an ecosystem prefix there, so only strip when

the leading segment cannot be one: without the guard,

``ignore @types/node@*`` would also silently ignore alerts for a package

literally named ``node``.

"""

package_names = {pkg_name}

if pkg_type:

package_names.add(f"{pkg_type}/{pkg_name}")

target_names = {name}

if not pkg_type and "/" in name and not name.startswith("@"):

target_names.add(name.split("/", 1)[1])

return bool(package_names & target_names) and (pkg_version == version or version == "*")

@staticmethod

def is_heading_line(line) -> bool:

is_heading_line = True

if line != "|Alert|Package|Introduced by|Manifest File|CI|" and ":---" not in line:

is_heading_line = False

return is_heading_line

@staticmethod

def extract_report_url(body: str) -> str:

"""

Pulls the Socket report link out of an existing comment body so it can be

carried over when the comment is rewritten.

:param body: str - The existing comment body.

:return: str - The report URL without its query string, or "" if absent.

"""

match = Comments.VIEW_REPORT_PATTERN.search(body)

if not match:

return ""

return match.group(1).split("?", 1)[0]

@staticmethod

def process_security_comment(comment: Comment, comments) -> str:

ignore_all, ignore_commands = Comments.get_ignore_options(comments)

if "start-socket-alerts-table" in "".join(comment.body_list):

new_body = Comments.process_original_security_comment(comment, ignore_all, ignore_commands)

else:

new_body = Comments.process_updated_security_comment(comment, ignore_all, ignore_commands)

return new_body

@staticmethod

def parse_alert_table_row(line: str) -> Optional[tuple[str, str, str]]:

"""Pull ``(ecosystem, package, version)`` out of a legacy alert table row.

Returns None for any row that does not have the expected shape rather than

raising. The row comes back from the provider's API, so its contents are

outside this process's control. Malformed cells must not interrupt status

reporting. A row that cannot be read is a row whose alert stays reported.

"""

cells = line.strip().lstrip("|").rstrip("|").split("|")

if len(cells) != 5:

return None

package = cells[1]

if "](" not in package:

return None

details = package.split("](", 1)[0].lstrip("[")

if "/" not in details:

return None

ecosystem, remainder = details.split("/", 1)

if "@" not in remainder:

return None

# Split from the right: a scoped name carries its own "@".

pkg_name, pkg_version = remainder.rsplit("@", 1)

if not pkg_name or not pkg_version:

return None

return ecosystem, pkg_name, pkg_version

@staticmethod

def process_original_security_comment(

comment: Comment,

ignore_all: bool,

ignore_commands: list[tuple[str, str]]

) -> str:

start = False

lines = []

kept_alert = False

for line in comment.body_list:

line = line.strip()

if "start-socket-alerts-table" in line:

start = True

lines.append(line)

elif start and "end-socket-alerts-table" not in line and not Comments.is_heading_line(line) and line != '':

parsed = Comments.parse_alert_table_row(line)

# ignore_all has to be checked outside the loop: an ignore-all

# comment produces no ignore_commands, so a loop-internal check

# never runs and every row was kept.

if parsed is None:

# An unparseable row cannot be evaluated against the ignore

# commands, so keep it: leaving an alert reported is the safe

# direction, and the comment body is not ours to discard.

ignore = ignore_all

else:

ecosystem, pkg_name, pkg_version = parsed

ignore = ignore_all or any(

Comments.is_ignore(pkg_name, pkg_version, name, version, ecosystem)

for name, version in ignore_commands

)

if not ignore:

kept_alert = True

lines.append(line)

elif "end-socket-alerts-table" in line:

start = False

lines.append(line)

else:

lines.append(line)

if not kept_alert:

return Messages.security_comment_no_alerts_template(

Comments.extract_report_url("\n".join(comment.body_list))

)

return "\n".join(lines)

@staticmethod

def process_updated_security_comment(

comment: Comment,

ignore_all: bool,

ignore_commands: list[tuple[str, str]]

) -> str:

"""

Processes an updated security comment containing an HTML table with alert sections.

Removes entire sections marked by start and end hidden comments if the alert matches

ignore conditions.

:param comment: Comment - The raw comment object containing the existing information.

:param ignore_all: bool - Flag to ignore all alerts.

:param ignore_commands: list of tuples - Specific ignore commands representing (pkg_name, pkg_version).

:return: str - The updated comment as a single string.

"""

lines = []

ignore_section = False

kept_alert = False # Whether any alert row survived the ignore commands

pkg_name = pkg_version = "" # Track current package and version

# Loop through the comment lines

for line in comment.body_list:

# Match on the stripped line but keep the original, so the markup is

# rewritten with the same indentation it was generated with.

line = line.rstrip("\r")

stripped = line.strip()

# Detect the start of an alert section

if stripped.startswith("<!-- start-socket-alert-"):

# Extract package name and version from the comment

try:

start_marker = stripped[len("<!-- start-socket-alert-"):-4] # Strip the comment markers

pkg_name, pkg_version = start_marker.rsplit("@", 1)

except ValueError:

pkg_name, pkg_version = "", ""

# Determine if we should ignore this alert

ignore_section = ignore_all or any(

Comments.is_ignore(pkg_name, pkg_version, name, version)

for name, version in ignore_commands

)

# If not ignored, include this start marker

if not ignore_section:

kept_alert = True

lines.append(line)

# Detect the end of an alert section

elif stripped.startswith("<!-- end-socket-alert-"):

# Only include if we are not ignoring this section

if not ignore_section:

lines.append(line)

ignore_section = False # Reset ignore flag

# Include lines inside an alert section only if not ignored

elif not ignore_section:

lines.append(line)

# Every row was ignored, so drop the table rather than leaving the caution

# banner sitting above an empty one.

if not kept_alert:

return Messages.security_comment_no_alerts_template(

Comments.extract_report_url("\n".join(comment.body_list))

)

return Messages.normalize_comment_html("\n".join(lines))

@staticmethod

def extract_alert_details_from_row(row: str, ignore_all: bool, ignore_commands: list[tuple[str, str]]) -> tuple:

"""

Parses an HTML table row (<tr>) to extract alert details and determine if it should be ignored.

:param row: str - The HTML table row as a string.

:param ignore_all: bool - Flag to ignore all alerts.

:param ignore_commands: list of tuples - List of (pkg_name, pkg_version) to ignore.

:return: tuple - (pkg_name, pkg_version, ignore)

"""

# Extract package details (pkg_name and pkg_version) from the HTML table row

try:

# Find the relevant <summary> element to extract package information

start_index = row.index("<summary>")

end_index = row.index("</summary>")

summary_content = row[start_index + 9:end_index] # Extract content between <summary> tags

# Example: "npm/malicious-package@1.0.0 - Known Malware Alert"

pkg_info, _ = summary_content.split(" - ", 1)

pkg_name, pkg_version = pkg_info.split("@")

except ValueError:

# If parsing fails, skip this row

return "", "", False

# Check ignore logic

ignore = False

for name, version in ignore_commands:

if ignore_all or Comments.is_ignore(pkg_name, pkg_version, name, version):

ignore = True

break

return pkg_name, pkg_version, ignore

@staticmethod

def check_for_socket_comments(

comments: dict,

is_authorized: Optional[Callable[[Comment], bool]] = None

):

"""Bucket a pull request's comments into the ones the CLI acts on.

``is_authorized`` gates the ignore bucket, and is the only place that gate

exists: an ``@SocketSecurity ignore`` command suppresses a security alert,

so it is honored only from someone with write access to the repository.

Filtering here rather than at each consumer means the rejected command is

also absent from the ignore telemetry, which should record what was acted

on. Both SCM adapters supply a predicate; omitting it trusts every

commenter and is only appropriate in tests.

"""

socket_comments = {}

for comment_id in comments:

comment = comments[comment_id]

comment: Comment

if "socket-security-comment-actions" in comment.body:

socket_comments["security"] = comment

elif "socket-overview-comment-actions" in comment.body:

socket_comments["overview"] = comment

elif "SocketSecurity ignore".lower() in comment.body_list[0].lower():

if is_authorized is not None and not is_authorized(comment):

log.warning(

"Skipping @SocketSecurity ignore command from "

f"{Comments.comment_author_name(comment)}: no write access "

"to this repository. Alerts remain reported."

)

continue

if "ignore" not in socket_comments:

socket_comments["ignore"] = []

socket_comments["ignore"].append(comment)

return socket_comments