Skip to content

Latest commit

 

History

History
206 lines (141 loc) · 6.82 KB

15.api.md

File metadata and controls

206 lines (141 loc) · 6.82 KB
Raw

API interface

levrt.Cr

  • Construction method

    levrt.Cr(image, schema, host, entry, cmd, env, files, links, ports, services)

  • Interface description:：

    Define a docker image object and return. All subsequent tasks are executed in this image.

  • Parameter list

    Parameter name Type Is it mandatory Illustration default Example
    image str Yes Image name None levkit/nmap_python:v1.0
    schema str No The definition name of the data to be stored, the format is<namespace>/<name>@<major>.<minor>. None talentsec/[email protected]
    host boolean No Host network mode False
    entry method No Entry function call None
    cmd list[str] No Command line parameters [] ["--option", "-f"]
    env list[str] No Environment variable [] ["PATH=/bin", "DB_URI=..."]
    files dict[ str, File ] No Mount file, key: target absolute path, value：file content or local file path, {} {"/etc/config.json": json.dumps(...).encode(),"/data/blob": pathlib.Path("./local_path...")}
    links dict[ str, str ] No Host link alias {} {"proxy": other_task.host}
    ports list[int] No Wait for the tcp port to be available [] [8000]
    services dict[str, Cr] No Dependent services, started before the main container runs and destroyed when the main container ends {} {"redis", Cr("redis:latest")}

  • Method

    class Cr:
    async def start(self) -> Task:
        """
        Start a container and return the corresponding task.
        """
        pass

    async def done(self) -> Document:
        """
        Start a container, wait for the run to complete. It will return a document of its result data.
        """
        pass

    def __await__(self):
        return self.done().__await__()

levrt.Task

  • Construction method

    None

  • method

    class Task:
    async def signal(self, signal="SIGINT"):
        """
        Send Unix signals to the main container.
        """
        pass

    async def done(self) -> Document:
        """
        Wait for the main container to finish running.
        """
        pass

    def __await__(self):
        return self.done().__await__()

levrt.Document

  • Construction method

    None

  • attribute

    • coll:

      mongodb collection

    • __filters：

      dict ， _id field of stored data

  • method

    class Document:
    async def get(self, query=None, projection=None):
        """
        Use __filters to get the corresponding document,
        Basically equivalent to the mongodb query statement: db.collection.findOne(query, projection)
        Get the dataset in document
        """
        pass

levrt.lev

  • Construction method

    None

  • role

    Store data.

  • method

    # Store data.
def set(self, data=None, **kwargs)

  • Example:

    import levrt
from levrt import lev


def scan(target: str):
    @levrt.remote
    def entry():
        issues=["Some issues"]
        # Store data.
        lev.set(target=target, issues=issues)
    
    return Cr("...", "ts/[email protected]", entry=entry())


async def main():
    doc = await scan("example.com")
    assert {"target": "example.com", "issues": ["Some issues"]} == await doc.get()
    assert doc.coll.collaction_name == "ts/[email protected]"
    # stored in ts/[email protected]

    with lev / "example":
        example = await scan("example.com")
        # stored in example|ts/[email protected]

    with lev / "httpbin" / "org":
        httpbin = await scan("httpbin.org")
        # stored in httpbin|org|ts/[email protected]

levrt.Concurrent

  • Construction method

    None

  • role

    Multithreaded execution task

  • method

    class Concurrent:
    def __init__(self, limit=None)

    def start(self, coroutine) -> asyncio.Task

  • Example

    async def current():
    urls = [...]

    # Multiple scan tasks will run at the same time, but at most 5.
    async with Concurrent(limit=5) as c:
        for url in urls:
            c.start(scan(url))

annot.meta

  • Construction method annot.meta(exports, desc, params,cats,tags,ext)

  • role

    Used for descriptions of patterns/workflow functions, or exported modules.

  • Parameter description

    Parameter name Type Is it mandatory Illustration default Example
    export list No Current module exports None [raw, alive, port_os]
    desc str No Description ‘’ ”detect surviving hosts in LAN“
    params list No Parameter list [] [annot.Param("ip", "target ip to scan", holder="192.168.1.1/24")]
    cats list | dict No Module or workflow category (refer to 14. FAQ appendix for details) None {
    Attck: [Attck.Reconnaissance],
    BlackArch: [BlackArch.Scanner]
    }
    tags list[str] No Tool mode or workflow tags for searching [] [ 'nmap', 'scan' ]

  • Example

    Reference: 6. Build the security tool nmap code