Merge pull request #13 from Veridise:nikos/projects-work

Nikos/projects-work

Author: NikosChondros

audithub_client/__main__.py

@@ -1,13 +1,10 @@
-import logging
-import sys
 from typing import Annotated, Literal
 
 from cyclopts import Group, Parameter
 
-from .library.context import AuditHubContext
-from .library.invocation_common import app
+from audithub_client.library.logging_level import set_logging_level
 
-app.meta.group_parameters = Group("Global Parameters")
+from .library.invocation_common import app
 from .scripts.create_version_via_local_archive import (  # noqa
     create_version_via_local_archive,
 )
@@ -23,6 +20,7 @@
 from .scripts.get_users import get_users  # noqa
 from .scripts.get_version_archive import get_version_archive  # noqa
 from .scripts.monitor_task import monitor_task  # noqa
+from .scripts.qa_cleanup import qa_cleanup  # noqa
 from .scripts.start_orca_task import start_orca_task  # noqa
 from .scripts.start_picus_v2_task import start_picus_v2_task  # noqa
 from .scripts.start_vanguard_task import (  # noqa
@@ -31,45 +29,12 @@
     start_zk_vanguard_task,
 )
 
-
-class LevelFormatter(logging.Formatter):
-    def __init__(self, formats, default_fmt=None):
-        super().__init__()
-        self.formats = formats
-        self.default_fmt = default_fmt or "%(levelname)s: %(message)s"
-
-    def format(self, record):
-        fmt = self.formats.get(record.levelno, self.default_fmt)
-        formatter = logging.Formatter(fmt)
-        return formatter.format(record)
+app.meta.group_parameters = Group("Global Parameters", sort_key=99)
 
 
 @app.meta.default
 def meta(
     *tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
-    base_url: Annotated[
-        str, Parameter(env_var="AUDITHUB_BASE_URL", help="AuditHub base URL")
-    ],
-    oidc_configuration_url: Annotated[
-        str,
-        Parameter(
-            env_var="AUDITHUB_OIDC_CONFIGURATION_URL",
-            help="AuditHub OpenID Connect configuration URL",
-        ),
-    ],
-    oidc_client_id: Annotated[
-        str,
-        Parameter(
-            env_var="AUDITHUB_OIDC_CLIENT_ID", help="AuditHub OpenID Connect client id"
-        ),
-    ],
-    oidc_client_secret: Annotated[
-        str,
-        Parameter(
-            env_var="AUDITHUB_OIDC_CLIENT_SECRET",
-            help="AuditHub OpenID Connect client secret. Please note that this is confidential information.",
-        ),
-    ],
     log_level: Annotated[
         Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
         Parameter(
@@ -79,49 +44,11 @@ def meta(
         ),
     ] = "INFO",
 ):
-    # Leave all levels except DEBUG "simple", without outputting the module name.
-    # At debug level, also show the module name that produces the log message.
-    default_log_format = "%(asctime)s %(levelname)s %(message)s"
-    debug_log_format = "%(asctime)s %(levelname)s %(name)s %(message)s"
-
-    # Setup logging as usual
-    logging.basicConfig(
-        level=log_level,
-        format=default_log_format,
-        datefmt="%H:%M:%S",
-        stream=sys.stderr,
-    )
-    # Now modify the root logger
-    root_logger = logging.getLogger()
-    handler = root_logger.handlers[0]  # basicConfig creates one handler by default
-    handler.setFormatter(
-        LevelFormatter(
-            {logging.DEBUG: debug_log_format}, default_fmt=default_log_format
-        )
-    )
-
-    # For these modules, raise the log level to reduce noise
-    for module in ["httpx", "httpcore"]:
-        module_logger = logging.getLogger(module)
-        module_logger.setLevel(logging.WARNING)
+    set_logging_level(log_level)
 
     command, bound, _ignored = app.parse_args(tokens)
-    # When this script runs with no args, help_print is automatically invoked
-    # Only in this situation however, it fails with: "TypeError: App.help_print() got an unexpected keyword argument 'rpc_context'"
-    # By only, I mean it does not fail when invoked with "--help".
-    # So let's treat it differently
-    if command == app.help_print:
-        return command(*bound.args, **bound.kwargs)
-    return command(
-        *bound.args,
-        **bound.kwargs,
-        rpc_context=AuditHubContext(
-            base_url=base_url,
-            oidc_configuration_url=oidc_configuration_url,
-            oidc_client_id=oidc_client_id,
-            oidc_client_secret=oidc_client_secret,
-        ),
-    )
+
+    return command(*bound.args, **bound.kwargs)
 
 
 def main():

audithub_client/api/create_project.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+from dataclasses import dataclass
+
+from ..library.auth import authentication_retry
+from ..library.context import AuditHubContext
+from ..library.http import POST
+from ..library.net_utils import ensure_success, response_json
+from .models import ProjectInfo
+
+
+@dataclass
+class CreateProjectArgs:
+    organization_id: int
+    project_definition: ProjectInfo
+    temp_version_id: int | None
+
+
+def api_create_project(context: AuditHubContext, input: CreateProjectArgs) -> dict:
+    params = {}
+    if input.temp_version_id:
+        params["temp_version_id"] = input.temp_version_id
+    response = authentication_retry(
+        context,
+        POST,
+        url=f"{context.base_url}/organizations/{input.organization_id}/projects",
+        json=input.project_definition.model_dump(),
+        params=params,
+    )
+    ensure_success(response)
+    ret = response_json(response)
+    return ret

audithub_client/api/create_temp_version_via_local_archive.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python3
+from dataclasses import dataclass
+
+from ..library.auth import authentication_retry
+from ..library.context import AuditHubContext
+from ..library.http import POST
+from ..library.net_utils import ensure_success, response_json
+
+
+@dataclass
+class CreateTempVersionViaLocalArchiveArgs:
+    organization_id: int
+    name: str
+
+
+def api_create_temp_version_via_local_archive(
+    context: AuditHubContext, input: CreateTempVersionViaLocalArchiveArgs, fp
+):
+    data = {"name": input.name}
+
+    response = authentication_retry(
+        context,
+        POST,
+        url=f"{context.base_url}/organizations/{input.organization_id}/temp-versions",
+        data=data,
+        files={"archive": ("sources.zip", fp, "application/zip")},
+    )
+    ensure_success(response)
+    ret = response_json(response)
+    return ret

audithub_client/api/create_temp_version_via_url.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+from dataclasses import dataclass
+from typing import Literal, Optional
+
+from ..library.auth import authentication_retry
+from ..library.context import AuditHubContext
+from ..library.http import POST
+from ..library.net_utils import ensure_success, response_json
+from ..library.utils import get_dict_of_fields_except
+
+
+@dataclass
+class CreateTempVersionViaUrlArgs:
+    organization_id: int
+    name: str
+    input_type: Literal["git", "archive"]
+    url: str
+    revision: Optional[str] = None
+    includes_submodules: Optional[bool] = None
+
+
+def api_create_temp_version_via_url(
+    context: AuditHubContext, input: CreateTempVersionViaUrlArgs
+):
+    data = get_dict_of_fields_except(input, {"organization_id"})
+
+    response = authentication_retry(
+        context,
+        POST,
+        url=f"{context.base_url}/organizations/{input.organization_id}/temp-versions-url",
+        data=data,
+    )
+    ensure_success(response)
+    ret = response_json(response)
+    return ret

audithub_client/api/get_configuration.py

@@ -3,9 +3,11 @@
 from ..library.context import AuditHubContext
 from ..library.http import get
 from ..library.net_utils import ensure_success, response_json
+from ..library.ssl import get_verify_ssl
 
 
 def api_get_configuration(context: AuditHubContext):
-    response = get(url=f"{context.base_url}/configuration")
+    url = f"{context.base_url}/configuration"
+    response = get(url=url, verify=get_verify_ssl(url))
     ensure_success(response)
     return response_json(response)

audithub_client/api/get_latest_version.py

@@ -0,0 +1,26 @@
+#!/usr/bin/env python3
+from dataclasses import dataclass
+
+from ..library.auth import authentication_retry
+from ..library.context import AuditHubContext
+from ..library.http import GET
+from ..library.net_utils import ensure_success, response_json
+
+
+@dataclass
+class GetLatestVersionArgs:
+    organization_id: int
+    project_id: int
+
+
+def api_get_latest_version(
+    context: AuditHubContext, input: GetLatestVersionArgs
+) -> dict:
+    response = authentication_retry(
+        context,
+        GET,
+        url=f"{context.base_url}/organizations/{input.organization_id}/projects/{input.project_id}/versions/latest",
+    )
+    ensure_success(response)
+    ret = response_json(response)
+    return ret

audithub_client/api/get_project.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+from dataclasses import dataclass
+
+from ..library.auth import authentication_retry
+from ..library.context import AuditHubContext
+from ..library.http import GET
+from ..library.net_utils import ensure_success, response_json
+
+
+@dataclass
+class GetProjectArgs:
+    organization_id: int
+    project_id: int
+
+
+def api_get_project(context: AuditHubContext, input: GetProjectArgs) -> dict:
+    response = authentication_retry(
+        context,
+        GET,
+        url=f"{context.base_url}/organizations/{input.organization_id}/projects/{input.project_id}",
+    )
+    ensure_success(response)
+    ret = response_json(response)
+    return ret

audithub_client/api/models/__init__.py

@@ -0,0 +1,77 @@
+from datetime import datetime
+from typing import Literal, Optional
+
+from pydantic import BaseModel, Field, RootModel
+
+
+class EnvVar(BaseModel):
+    name: str
+    value: str
+
+
+class GitInput(BaseModel):
+    url: str
+    input_type: Literal["git"] = "git"
+    includes_submodules: bool = False
+    revision: Optional[str] = None
+
+
+class ArchiveInput(BaseModel):
+    input_type: Literal["archive"] = "archive"
+    url: Optional[str] = None
+
+
+class NPMProjectDependency(BaseModel):
+    tool: Literal["npm", "yarn", "pnpm"]
+    lockfile: bool
+    node_version: Optional[str] = None
+
+
+class ProjectDependency(BaseModel):
+    npm: Optional[NPMProjectDependency] = None
+    foundry: Optional[bool] = False
+
+
+class ProjectInfo(BaseModel):
+    name: str
+    project_root: str
+    env_vars: list[EnvVar] = Field(default_factory=list)
+    dependencies: Optional[ProjectDependency] = None
+    build_system: Optional[Literal["hardhat", "hardhat-ignition", "foundry"]] = None
+    contents: Optional[list[Literal["circom", "solidity", "picus", "llzk"]]] = Field(
+        default_factory=list
+    )
+    src_path: str
+    include_path: Optional[str] = None
+    specs_path: Optional[str] = None
+    hints_path: Optional[str] = None
+    deployment_script_path: Optional[str] = None
+    input_info: GitInput | ArchiveInput
+
+
+class Project(ProjectInfo):
+    id: int
+    created_at: datetime
+
+
+class IdAndMessageResponse(BaseModel):
+    id: int
+    message: str
+
+
+class UserProfile(BaseModel):
+    id: str
+    name: str
+    email: str
+    rights: list[str]
+    disable_online_notifications: bool
+    disable_digest_notifications: bool
+
+
+class OrganizationAccessRestriction(BaseModel):
+    function: str
+    value: Optional[int] = None
+    detector: Optional[str] = None
+
+
+OrganizationAccessRestrictions = RootModel[list[OrganizationAccessRestriction]]

audithub_client/api/monitor_task.py

@@ -6,6 +6,8 @@
 from websockets import ConnectionClosedOK
 from websockets.sync.client import connect
 
+from audithub_client.library.ssl import get_websocket_ssl_context
+
 from ..library.auth import get_access_token, get_token_header
 from ..library.context import AuditHubContext
 
@@ -39,7 +41,9 @@ def api_monitor_task(context: AuditHubContext, input: MonitorTaskArgs) -> bool:
         f"{url}/organizations/{input.organization_id}/tasks/{input.task_id}/progress"
     )
     status: ALLOWED_STATUS_OPTIONS
-    with connect(endpoint, max_size=100 * 1024 * 1024) as websocket:
+    with connect(
+        endpoint, max_size=100 * 1024 * 1024, **get_websocket_ssl_context(url)
+    ) as websocket:
         try:
             # send initial authentication
             websocket.send(token)