Skip to content

osco_result_to_oscal_ar

trestle.tasks.osco_result_to_oscal_ar ¤

OSCAL transformation tasks.

Attributes¤

logger = logging.getLogger(__name__) module-attribute ¤

Classes¤

OscoResultToOscalAR ¤

Bases: TaskBase

Task to convert Osco result to OSCAL json.

Attributes:

Name Type Description
name

Name of the task.
Source code in trestle/tasks/osco_result_to_oscal_ar.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class OscoResultToOscalAR(TaskBase):
    """
    Task to convert Osco result to OSCAL json.

    Attributes:
        name: Name of the task.
    """

    name = 'osco-result-to-oscal-ar'

    def __init__(self, config_object: Optional[configparser.SectionProxy]) -> None:
        """
        Initialize trestle task osco-result-to-oscal-ar.

        Args:
            config_object: Config section associated with the task.
        """
        super().__init__(config_object)

    def print_info(self) -> None:
        """Print the help string."""
        logger.info(f'Help information for {self.name} task.')
        logger.info('')
        logger.info(
            'Purpose: Transform Osco files into Open Security Controls Assessment Language (OSCAL) '
            + 'partial results files.'
        )
        logger.info('')
        logger.info('Configuration flags sit under [task.osco-result-to-oscal-ar]:')
        logger.info(
            '  checking  = (optional) True indicates perform strict checking of OSCAL properties, default is False.'
        )
        logger.info('  input-dir = (required) the path of the input directory comprising Osco results.')
        logger.info(
            '  output-dir = (required) the path of the output directory comprising synthesized OSCAL .json files.'
        )
        logger.info('  output-overwrite = (optional) true [default] or false; replace existing output when true.')
        logger.info(
            '  quiet = (optional) true or false [default]; display file creations and rules analysis when false.'
        )
        logger.info(
            '  timestamp = (optional) timestamp for the Observations in ISO 8601 format, such as '
            + ' 2021-01-04T00:05:23+04:00 for example; if not specified then value for "Timestamp" key in the Osco '
            + ' result is used if present, otherwise current time is used.'
        )
        logger.info('')
        logger.info(
            'Operation: A transformation is performed on one or more Osco input files to produce output in OSCAL '
            + 'partial results format.'
        )

    def simulate(self) -> TaskOutcome:
        """Provide a simulated outcome."""
        self._simulate = True
        return self._transform()

    def execute(self) -> TaskOutcome:
        """Provide an actual outcome."""
        self._simulate = False
        return self._transform()

    def _transform(self) -> TaskOutcome:
        """Perform transformation."""
        try:
            return self._transform_work()
        except Exception:
            logger.debug(traceback.format_exc())
            mode = ''
            if self._simulate:
                mode = 'simulated-'
            return TaskOutcome(mode + 'failure')

    def _transform_work(self) -> TaskOutcome:
        """
        Perform transformation work steps.

        Work steps: read input, process, write output, display analysis
        """
        mode = ''
        if self._simulate:
            mode = 'simulated-'
        if not self._config:
            logger.warning('config missing')
            return TaskOutcome(mode + 'failure')
        # config required input & output dirs
        try:
            idir = self._config['input-dir']
            ipth = pathlib.Path(idir)
            odir = self._config['output-dir']
            opth = pathlib.Path(odir)
        except KeyError as e:
            logger.debug(f'key {e.args[0]} missing')
            return TaskOutcome(mode + 'failure')
        # config optional overwrite & quiet
        self._overwrite = self._config.getboolean('output-overwrite', True)
        quiet = self._config.get('quiet', False)
        self._verbose = not self._simulate and not quiet
        # config optional timestamp
        timestamp = self._config.get('timestamp')
        if timestamp is not None:
            try:
                OscoTransformer.set_timestamp(timestamp)
            except Exception:
                logger.warning('config invalid "timestamp"')
                return TaskOutcome(mode + 'failure')
        # config optional performance
        modes = {
            'checking': self._config.getboolean('checking', False),
        }
        # insure output dir exists
        opth.mkdir(exist_ok=True, parents=True)
        # process
        for ifile in sorted(ipth.iterdir()):
            if ifile.suffix not in ['.json', '.jsn', '.yaml', '.yml', '.xml']:
                continue
            blob = self._read_file(ifile)
            osco_transformer = OscoTransformer()
            osco_transformer.set_modes(modes)
            results = osco_transformer.transform(blob)
            oname = ifile.stem + '.oscal' + '.json'
            ofile = opth / oname
            if not self._overwrite and pathlib.Path(ofile).exists():
                logger.warning(f'output: {ofile} already exists')
                return TaskOutcome(mode + 'failure')
            self._write_file(results, ofile)
            self._show_analysis(osco_transformer)
        return TaskOutcome(mode + 'success')

    def _read_file(self, ifile: str) -> str:
        """Read raw input file."""
        if not self._simulate and self._verbose:
            logger.info(f'input: {ifile}')
        with open(ifile, encoding=const.FILE_ENCODING) as fp:
            blob = fp.read()
        return blob

    def _write_file(self, result: str, ofile: str) -> None:
        """Write oscal results file."""
        if not self._simulate:
            if self._verbose:
                logger.info(f'output: {ofile}')
            result.oscal_write(pathlib.Path(ofile))

    def _show_analysis(self, osco_transformer: OscoTransformer) -> None:
        """Show analysis."""
        if not self._simulate and self._verbose:
            analysis = osco_transformer.analysis
            for line in analysis:
                logger.info(line)
Attributes¤
name = 'osco-result-to-oscal-ar' class-attribute instance-attribute ¤
Functions¤
__init__(config_object) ¤

Initialize trestle task osco-result-to-oscal-ar.

Parameters:

Name Type Description Default
config_object Optional[SectionProxy]

Config section associated with the task.

 required
Source code in trestle/tasks/osco_result_to_oscal_ar.py 
42
43
44
45
46
47
48
49
def __init__(self, config_object: Optional[configparser.SectionProxy]) -> None:
    """
    Initialize trestle task osco-result-to-oscal-ar.

    Args:
        config_object: Config section associated with the task.
    """
    super().__init__(config_object)
execute() ¤

Provide an actual outcome.

Source code in trestle/tasks/osco_result_to_oscal_ar.py 
88
89
90
91
def execute(self) -> TaskOutcome:
    """Provide an actual outcome."""
    self._simulate = False
    return self._transform()
print_info() ¤

Print the help string.

Source code in trestle/tasks/osco_result_to_oscal_ar.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def print_info(self) -> None:
    """Print the help string."""
    logger.info(f'Help information for {self.name} task.')
    logger.info('')
    logger.info(
        'Purpose: Transform Osco files into Open Security Controls Assessment Language (OSCAL) '
        + 'partial results files.'
    )
    logger.info('')
    logger.info('Configuration flags sit under [task.osco-result-to-oscal-ar]:')
    logger.info(
        '  checking  = (optional) True indicates perform strict checking of OSCAL properties, default is False.'
    )
    logger.info('  input-dir = (required) the path of the input directory comprising Osco results.')
    logger.info(
        '  output-dir = (required) the path of the output directory comprising synthesized OSCAL .json files.'
    )
    logger.info('  output-overwrite = (optional) true [default] or false; replace existing output when true.')
    logger.info(
        '  quiet = (optional) true or false [default]; display file creations and rules analysis when false.'
    )
    logger.info(
        '  timestamp = (optional) timestamp for the Observations in ISO 8601 format, such as '
        + ' 2021-01-04T00:05:23+04:00 for example; if not specified then value for "Timestamp" key in the Osco '
        + ' result is used if present, otherwise current time is used.'
    )
    logger.info('')
    logger.info(
        'Operation: A transformation is performed on one or more Osco input files to produce output in OSCAL '
        + 'partial results format.'
    )
simulate() ¤

Provide a simulated outcome.

Source code in trestle/tasks/osco_result_to_oscal_ar.py 
83
84
85
86
def simulate(self) -> TaskOutcome:
    """Provide a simulated outcome."""
    self._simulate = True
    return self._transform()

handler: python