Files
trailofbits-buttercup/program-model/src/buttercup/program_model/__cli__.py
T
2over12 ea2fd093d5 Program model integrate (#13)
* Add documentation

* Add janusgraph custom configuration. Simply graph creation.

* Add properties to edges. Index on uri property.

* reformat

* Create libpng graph in a more reasonable amount of time.

* Update README.md

* Wrap indexing into function

* updating...

* update proto

* fix command

* Update settings for serve and processing indexing

* Add dockerfile to docker container

* Add trigger script

* Fix error with loading graphml file

* reformat

* Remove resource constraints for program model

* Change permissions of temp dir

* Update readme

* Update dockerignore

---------

Co-authored-by: Evan Downing <2077950+evandowning@users.noreply.github.com>
2025-02-18 11:59:16 -05:00

61 lines
1.9 KiB
Python

import logging
from buttercup.program_model.program_model import ProgramModel
from buttercup.program_model.settings import (
ProgramModelSettings,
ProgramModelServeCommand,
ProgramModelProcessCommand,
)
from buttercup.common.logger import setup_package_logger
from pydantic_settings import get_subcommand
from buttercup.common.datastructures.msg_pb2 import IndexRequest
from redis import Redis
logger = logging.getLogger(__name__)
def prepare_task(command: ProgramModelProcessCommand) -> IndexRequest:
"""Prepares task for indexing."""
return IndexRequest(
package_name=command.package_name,
sanitizer=command.sanitizer,
ossfuzz=command.ossfuzz,
source_path=command.source_path,
task_id=command.task_id,
build_type=command.build_type,
)
def main():
settings = ProgramModelSettings()
setup_package_logger(__name__, settings.log_level)
command = get_subcommand(settings)
if isinstance(command, ProgramModelServeCommand):
redis = Redis.from_url(command.redis_url, decode_responses=False)
with ProgramModel(
sleep_time=command.sleep_time,
redis=redis,
wdir=command.wdir,
script_dir=command.script_dir,
kythe_dir=command.kythe_dir,
python=command.python,
allow_pull=command.allow_pull,
base_image_url=command.base_image_url,
) as program_model:
program_model.serve()
elif isinstance(command, ProgramModelProcessCommand):
task = prepare_task(command)
with ProgramModel(
wdir=command.wdir,
script_dir=command.script_dir,
kythe_dir=command.kythe_dir,
python=command.python,
allow_pull=command.allow_pull,
base_image_url=command.base_image_url,
) as program_model:
program_model.process_task(task)
if __name__ == "__main__":
main()