pyt/pyt/vulnerabilities.py at 8githubScanRefactor · python-security/pyt · GitHub

Repository navigation

History

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

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

"""Module for finding vulnerabilities based on a definitions file."""

import ast

from collections import namedtuple

from .base_cfg import Node, AssignmentNode, ReturnNode

from .framework_adaptor import TaintedNode

from .vulnerability_log import (

Vulnerability,

VulnerabilityLog,

SanitisedVulnerability

)

from .lattice import Lattice

from .vars_visitor import VarsVisitor

from .trigger_definitions_parser import default_trigger_word_file, parse

Triggers = namedtuple('Triggers', 'sources sinks sanitiser_dict')

Sanitiser = namedtuple('Sanitiser', 'trigger_word cfg_node')

class TriggerNode():

def __init__(self, trigger_word, sanitisers, cfg_node,

secondary_nodes=None):

self.trigger_word = trigger_word

self.sanitisers = sanitisers

self.cfg_node = cfg_node

self.secondary_nodes = secondary_nodes

def append(self, cfg_node):

if not cfg_node == self.cfg_node:

if self.secondary_nodes and cfg_node not in self.secondary_nodes:

self.secondary_nodes.append(cfg_node)

elif not self.secondary_nodes:

self.secondary_nodes = [cfg_node]

def identify_triggers(cfg, sources, sinks):

"""Identify sources, sinks and sanitisers in a CFG.

Args:

cfg(CFG): CFG to find sources, sinks and sanitisers in.

sources(tuple): list of sources,

a source is a (source, sanitiser) tuple.

sinks(tuple): list of sources, a sink is a (sink, sanitiser) tuple.

Returns:

Triggers tuple with sink and source nodes and a sanitiser node dict.

"""

assignment_nodes = filter_cfg_nodes(cfg, AssignmentNode)

tainted_nodes = filter_cfg_nodes(cfg, TaintedNode)

tainted_trigger_nodes = [TriggerNode('Flask function URL parameter', None,

node) for node in tainted_nodes]

sources_in_file = find_triggers(assignment_nodes, sources)

sources_in_file.extend(tainted_trigger_nodes)

find_secondary_sources(assignment_nodes, sources_in_file)

sinks_in_file = find_triggers(cfg.nodes, sinks)

sanitiser_node_dict = build_sanitiser_node_dict(cfg, sinks_in_file)

return Triggers(sources_in_file, sinks_in_file, sanitiser_node_dict)

def filter_cfg_nodes(cfg, cfg_node_type):

return [node for node in cfg.nodes if isinstance(node, cfg_node_type)]

def find_secondary_sources(assignment_nodes, sources):

assignments = dict()

for source in sources:

source.secondary_nodes = find_assignments(assignment_nodes, source)

def find_assignments(assignment_nodes, source):

old = list()

# added in order to propagate reassignments of the source node:

new = [source.cfg_node]

update_assignments(new, assignment_nodes, source.cfg_node)

while new != old:

old = new

update_assignments(new, assignment_nodes, source.cfg_node)

new.remove(source.cfg_node) # remove source node from result

return new

def update_assignments(l, assignment_nodes, source):

for node in assignment_nodes:

for other in l:

if node not in l:

append_if_reassigned(l, other, node)

def append_if_reassigned(l, secondary, node):

# maybe: secondary in node.new_constraint and

try:

if secondary.left_hand_side in node.right_hand_side_variables or\

secondary.left_hand_side == node.left_hand_side:

l.append(node)

except AttributeError:

print(secondary)

print('EXCEPT' + secondary)

exit(0)

def find_triggers(nodes, trigger_words):

"""Find triggers from the trigger_word_list in the cfg.

Args:

cfg(CFG): the CFG to find triggers in.

trigger_word_list(list[string]): list of trigger words to look for.

Returns:

List of found TriggerNodes

"""

l = list()

for node in nodes:

l.extend(iter(label_contains(node, trigger_words)))

return l

def label_contains(node, trigger_words):

"""Determine if node contains any of the trigger_words provided.

Args:

node(Node): CFG node to check.

trigger_words(list[string]): list of trigger words to look for.

Returns:

Iterable of TriggerNodes found. Can be multiple because multiple

trigger_words can be in one node.

"""

for trigger_word_tuple in trigger_words:

if trigger_word_tuple[0] in node.label:

trigger_word = trigger_word_tuple[0]

sanitisers = trigger_word_tuple[1]

yield TriggerNode(trigger_word, sanitisers, node)

def build_sanitiser_node_dict(cfg, sinks_in_file):

"""Build a dict of string -> TriggerNode pairs, where the string

is the sanitiser and the TriggerNode is a TriggerNode of the sanitiser.

Args:

cfg(CFG): cfg to traverse.

sinks_in_file(list[TriggerNode]): list of TriggerNodes contains

the sinks in the file

Returns:

A string -> TriggerNode dict.

"""

sanitisers = list()

for sink in sinks_in_file:

sanitisers.extend(sink.sanitisers)

sanitisers_in_file = list()

for sanitiser in sanitisers:

for cfg_node in cfg.nodes:

if sanitiser in cfg_node.label:

sanitisers_in_file.append(Sanitiser(sanitiser, cfg_node))

sanitiser_node_dict = dict()

for sanitiser in sanitisers:

sanitiser_node_dict[sanitiser] = list(find_sanitiser_nodes(sanitiser,

sanitisers_in_file))

return sanitiser_node_dict

def find_sanitiser_nodes(sanitiser, sanitisers_in_file):

"""Find nodes containing a particular sanitiser.

Args:

sanitiser(string): sanitiser to look for.

sanitisers_in_file(list[Node]): list of CFG nodes with the sanitiser.

Returns:

Iterable of sanitiser nodes.

"""

for sanitiser_tuple in sanitisers_in_file:

if sanitiser == sanitiser_tuple.trigger_word:

yield sanitiser_tuple.cfg_node

def is_sanitized(sink, sanitiser_dict, lattice):

"""Check if sink is sanitised by any santiser in the sanitiser_dict.

Args:

sink(TriggerNode): TriggerNode of the sink.

sanitiser_dict(dict): dictionary of sink sanitiser pairs.

Returns:

True of False

"""

for sanitiser in sink.sanitisers:

for cfg_node in sanitiser_dict[sanitiser]:

if lattice.in_constraint(cfg_node, sink.cfg_node):

return True

return False

class SinkArgsError(Exception):

pass

def get_sink_args(cfg_node):

vv = VarsVisitor()

vv.visit(cfg_node.ast_node)

return vv.result

def get_vulnerability(source, sink, triggers, lattice):

"""Get vulnerability between source and sink if it exists.

Uses triggers to find sanitisers

Args:

source(TriggerNode): TriggerNode of the source.

sink(TriggerNode): TriggerNode of the sink.

triggers(Triggers): Triggers of the CFG.

Returns:

A Vulnerability if it exists, else None

"""

source_in_sink = lattice.in_constraint(source.cfg_node, sink.cfg_node)

secondary_in_sink = []

if source.secondary_nodes:

secondary_in_sink = [secondary for secondary in source.secondary_nodes

if lattice.in_constraint(secondary,

sink.cfg_node)]

trigger_node_in_sink = source_in_sink or secondary_in_sink

sink_args = get_sink_args(sink.cfg_node)

source_lhs_in_sink_args = source.cfg_node.left_hand_side in sink_args\

if sink_args else None

secondary_nodes_in_sink_args = any(True for node in secondary_in_sink

if node.left_hand_side in sink_args)\

if sink_args else None

lhs_in_sink_args = source_lhs_in_sink_args or secondary_nodes_in_sink_args

if trigger_node_in_sink and lhs_in_sink_args:

source_trigger_word = source.trigger_word

sink_trigger_word = sink.trigger_word

sink_is_sanitised = is_sanitized(sink, triggers.sanitiser_dict,

lattice)

if not sink_is_sanitised:

return Vulnerability(source.cfg_node, source_trigger_word,

sink.cfg_node, sink_trigger_word,

source.secondary_nodes)

elif sink_is_sanitised:

return SanitisedVulnerability(source.cfg_node, source_trigger_word,

sink.cfg_node, sink_trigger_word,

sink.sanitisers,

source.secondary_nodes)

return None

def find_vulnerabilities_in_cfg(cfg, vulnerability_log, definitions, lattice):

"""Find vulnerabilities in a cfg.

Args:

cfg(CFG): The CFG look find vulnerabilities in.

vulnerabilitiy_log: The log in which to place found vulnerabilities.

definitions: Source and sink definitions.

"""

triggers = identify_triggers(cfg, definitions.sources, definitions.sinks)

for sink in triggers.sinks:

for source in triggers.sources:

vulnerability = get_vulnerability(source, sink, triggers, lattice)

if vulnerability:

vulnerability_log.append(vulnerability)

def find_vulnerabilities(cfg_list, analysis_type,

trigger_word_file=default_trigger_word_file):

"""Find vulnerabilities in a list of CFGs from a trigger_word_file.

Args:

cfg_list (list): the list of CFGs to scan.

trigger_word_file (string): file containing trigger words.

Defaults to the flask trigger word file.

Returns:

A VulnerabilityLog with found vulnerabilities.

"""

definitions = parse(trigger_word_file)

vulnerability_log = VulnerabilityLog()

for cfg in cfg_list:

find_vulnerabilities_in_cfg(cfg, vulnerability_log, definitions,

Lattice(cfg.nodes, analysis_type))

return vulnerability_log

pyt/pyt/vulnerabilities.py at 8githubScanRefactor · python-security/pyt · GitHub

Repository navigation

Expand file tree

Breadcrumbs

vulnerabilities.py

Latest commit

History

Breadcrumbs

vulnerabilities.py

File metadata and controls