<commit_before>#!/usr/bin/python

## Binary Analysis Tool
## Copyright 2009-2015 <NAME> for Tjaldur Software Governance Solutions
## Licensed under Apache 2.0, see LICENSE file for details

'''
This file contains a few methods that can be useful for security scanning.
'''

import os, sys, zipfile, subprocess, re, cPickle, copy, tempfile
import bat.batdb

## This method extracts the CRC32 checksums from the entries of the encrypted zip file and checks
## whether or not there are any files in the database with the same CRC32. If so, a known plaintext
## attack is possible to decrypt the archive and extract the key. The return value will be the files
## or checksums in the database for which there is a plaintext version available.
def scanEncryptedZip(path, tags, blacklist=[], scanenv={}, scandebug=False, unpacktempdir=None):
	if not 'zip' in tags and not 'encrypted' in tags:
		return

	if not scanenv.has_key('BAT_DB'):
		return

	encryptedzip = zipfile.ZipFile(path, 'r')
	encryptedinfos = encryptedzip.infolist()
	batdb = bat.batdb.BatDb(scanenv['DBBACKEND'])
	c = batdb.getConnection(scanenv['BAT_DB'],scanenv)
	cursor = c.cursor()
	plaintexts = set()
	query = batdb.getQuery("select sha256 from hashconversion where crc32=%s")
	for e in encryptedinfos:
		crc = e.CRC
		## if the CRC is 0 it is a directory entry
		if crc == 0:
			continue
		cursor.execute(query, (str(crc),))
		res = cursor.fetchone()
		if res != None:
			plaintexts.add(res[0])
	cursor.close()
	c.close()
	if len(plaintexts) != 0:
		return (['encryptedzip-attack'], plaintexts)
	return

def encryptedZipSetup(scanenv, debug=False):
	if not 'DBBACKEND' in scanenv:
		return (False, None)
	if scanenv['DBBACKEND'] == 'sqlite3':
		return encryptedZipSetup_sqlite3(scanenv, debug)
	elif scanenv['DBBACKEND'] == 'postgresql':
		return encryptedZipSetup_postgresql(scanenv, debug)
	return (False, None)

def encryptedZipSetup_postgresql(scanenv, debug=False):
	newenv = copy.deepcopy(scanenv)
	batdb = bat.batdb.BatDb('postgresql')
	conn = batdb.getConnection(None,scanenv)
        if conn == None:
		return (False, None)
	conn.close()
	return (True, newenv)

def encryptedZipSetup_sqlite3(scanenv, debug=False):
	## first check if there is a database defined
	if not scanenv.has_key('BAT_DB'):
		return (False, None)
	if not os.path.exists(scanenv['BAT_DB']):
		return (False, None)

	newenv = copy.deepcopy(scanenv)
	batdb = bat.batdb.BatDb(scanenv['DBBACKEND'])
	c = batdb.getConnection(scanenv['BAT_DB'])
	cursor = c.cursor()

	## then check the database schema to see if there are crc32 checksums
	res = c.execute("select * from sqlite_master where type='table' and name='hashconversion'").fetchall()
	if res == []:
		cursor.close()
		c.close()
		return (False, None)

	## then check if there is a column 'crc32'
	res = c.execute("pragma table_info('hashconversion')").fetchall()
	cursor.close()
	c.close()
	if res == []:
		return (False, None)

	process = False

	for i in res:
		if i[1] == 'crc32':
			process = True
			break
	if not process:
		return (False, None)
		
	return (True, newenv)

## experimental clamscan feature
## Always run freshclam before scanning to get the latest
## virus signatures!
def scanVirus(path, tags, blacklist=[], scanenv={}, scandebug=False, unpacktempdir=None):
	p = subprocess.Popen(['clamscan', "%s" % (path,)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
	(stanout, stanerr) = p.communicate()
	if p.returncode == 0:
               	return
	else:
		## Oooh, virus found!
		viruslines = stanout.split("\n")
		## first line contains the report:
		virusname = viruslines[0].strip()[len(path) + 2:-6]
		return (['virus'], virusname)

## experimental feature to detect possible smells in binaries
## Many ODMs use shell commands in their programs using the system() call
## which has turned out to be vulnerable in certain instances.
## Some of these can be detected by looking for typical shell invocation
## patterns, such as %s or * in combination with hard coded paths
## TODO: add more patterns
def scanShellInvocations(unpackreports, scantempdir, topleveldir, processors, scanenv, scandebug=False, unpacktempdir=None):
	for i in unpackreports:
		## Limit to ELF binaries for now
		if not unpackreports[i].has_key('tags'):
			continue
		if not unpackreports[i].has_key('checksum'):
			continue
		if not 'elf' in unpackreports[i]['tags']:
			continue
		if not 'identifier' in unpackreports[i]['tags']:
			continue

		filehash = unpackreports[i]['checksum']

		## read pickle file
		leaf_file = open(os.path.join(topleveldir, "filereports", "%s-filereport.pickle" % filehash), 'rb')
		leafreports = cPickle.load(leaf_file)
		leaf_file.close()

		strs = leafreports['identifier']['strings']
		buggylines = []
		for line in strs:
			if '/sbin' in line or '/bin' in line:
				possiblybuggy = False
				for c in ['%s', '*']:
					if c in line:
						buggylines.append(line)
						break
		## now write back the results
		if buggylines != []:
			leafreports['shellinvocations'] = buggylines
			leafreports['tags'].append('shellinvocations')
			unpackreports[i]['tags'].append('shellinvocations')
			leaf_file = open(os.path.join(topleveldir, "filereports", "%s-filereport.pickle" % filehash), 'wb')
			cPickle.dump(leafreports, leaf_file)
			leaf_file.close()

## stubs for cracking passwords with "<NAME>"
## 1. look for files called 'passwd' and 'shadow'
## 2. search for individual entries in the database
## 3. crack unknown entries
##
## This is implemented as an "aggregate scan". Running "<NAME>"
## is an expensive operation and often there can be duplicate password
## or shadow files in firmwares.
def crackPasswords(unpackreports, scantempdir, topleveldir, processors, scanenv, scandebug=False, unpacktempdir=None):
	passwdfiles = []
	for u in unpackreports.keys():
		if not (os.path.basename(u) == 'shadow' or os.path.basename(u) == 'passwd'):
			continue
		if 'symlink' in unpackreports[u]['tags']:
			continue
		if 'elf' in unpackreports[u]['tags']:
			continue
		if os.path.basename(u) == 'shadow':<commit_msg>add check for finding openssh private keys<commit_after>#!/usr/bin/python

## Binary Analysis Tool
## Copyright 2009-2015 <NAME> for Tjaldur Software Governance Solutions
## Licensed under Apache 2.0, see LICENSE file for details

'''
This file contains a few methods that can be useful for security scanning.
'''

import os, sys, zipfile, subprocess, re, cPickle, copy, tempfile
import bat.batdb

## This method extracts the CRC32 checksums from the entries of the encrypted zip file and checks
## whether or not there are any files in the database with the same CRC32. If so, a known plaintext
## attack is possible to decrypt the archive and extract the key. The return value will be the files
## or checksums in the database for which there is a plaintext version available.
def scanEncryptedZip(path, tags, blacklist=[], scanenv={}, scandebug=False, unpacktempdir=None):
	if not 'encrypted' in tags:
		return

	if not 'zip' in tags:
		return

	if not scanenv.has_key('BAT_DB'):
		return

	encryptedzip = zipfile.ZipFile(path, 'r')
	encryptedinfos = encryptedzip.infolist()
	batdb = bat.batdb.BatDb(scanenv['DBBACKEND'])
	c = batdb.getConnection(scanenv['BAT_DB'],scanenv)
	cursor = c.cursor()
	plaintexts = set()
	query = batdb.getQuery("select sha256 from hashconversion where crc32=%s")
	for e in encryptedinfos:
		crc = e.CRC
		## if the CRC is 0 it is a directory entry
		if crc == 0:
			continue
		cursor.execute(query, (str(crc),))
		res = cursor.fetchone()
		if res != None:
			plaintexts.add(res[0])
	cursor.close()
	c.close()
	if len(plaintexts) != 0:
		return (['encryptedzip-attack'], plaintexts)
	return

def encryptedZipSetup(scanenv, debug=False):
	if not 'DBBACKEND' in scanenv:
		return (False, None)
	if scanenv['DBBACKEND'] == 'sqlite3':
		return encryptedZipSetup_sqlite3(scanenv, debug)
	elif scanenv['DBBACKEND'] == 'postgresql':
		return encryptedZipSetup_postgresql(scanenv, debug)
	return (False, None)

def encryptedZipSetup_postgresql(scanenv, debug=False):
	newenv = copy.deepcopy(scanenv)
	batdb = bat.batdb.BatDb('postgresql')
	conn = batdb.getConnection(None,scanenv)
        if conn == None:
		return (False, None)
	conn.close()
	return (True, newenv)

def encryptedZipSetup_sqlite3(scanenv, debug=False):
	## first check if there is a database defined
	if not scanenv.has_key('BAT_DB'):
		return (False, None)
	if not os.path.exists(scanenv['BAT_DB']):
		return (False, None)

	newenv = copy.deepcopy(scanenv)
	batdb = bat.batdb.BatDb(scanenv['DBBACKEND'])
	c = batdb.getConnection(scanenv['BAT_DB'])
	cursor = c.cursor()

	## then check the database schema to see if there are crc32 checksums
	res = c.execute("select * from sqlite_master where type='table' and name='hashconversion'").fetchall()
	if res == []:
		cursor.close()
		c.close()
		return (False, None)

	## then check if there is a column 'crc32'
	res = c.execute("pragma table_info('hashconversion')").fetchall()
	cursor.close()
	c.close()
	if res == []:
		return (False, None)

	process = False

	for i in res:
		if i[1] == 'crc32':
			process = True
			break
	if not process:
		return (False, None)
		
	return (True, newenv)

## experimental clamscan feature
## Always run freshclam before scanning to get the latest
## virus signatures!
def scanVirus(path, tags, blacklist=[], scanenv={}, scandebug=False, unpacktempdir=None):
	p = subprocess.Popen(['clamscan', "%s" % (path,)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
	(stanout, stanerr) = p.communicate()
	if p.returncode == 0:
               	return
	else:
		## Oooh, virus found!
		viruslines = stanout.split("\n")
		## first line contains the report:
		virusname = viruslines[0].strip()[len(path) + 2:-6]
		return (['virus'], virusname)

## experimental feature to detect possible smells in binaries
## Many ODMs use shell commands in their programs using the system() call
## which has turned out to be vulnerable in certain instances.
## Some of these can be detected by looking for typical shell invocation
## patterns, such as %s or * in combination with hard coded paths
## TODO: add more patterns
def scanShellInvocations(unpackreports, scantempdir, topleveldir, processors, scanenv, scandebug=False, unpacktempdir=None):
	for i in unpackreports:
		## Limit to ELF binaries for now
		if not unpackreports[i].has_key('tags'):
			continue
		if not unpackreports[i].has_key('checksum'):
			continue
		if not 'elf' in unpackreports[i]['tags']:
			continue
		if not 'identifier' in unpackreports[i]['tags']:
			continue

		filehash = unpackreports[i]['checksum']

		## read pickle file
		leaf_file = open(os.path.join(topleveldir, "filereports", "%s-filereport.pickle" % filehash), 'rb')
		leafreports = cPickle.load(leaf_file)
		leaf_file.close()

		strs = leafreports['identifier']['strings']
		buggylines = []
		for line in strs:
			if '/sbin' in line or '/bin' in line:
				possiblybuggy = False
				for c in ['%s', '*']:
					if c in line:
						buggylines.append(line)
						break
		## now write back the results
		if buggylines != []:
			leafreports['shellinvocations'] = buggylines
			leafreports['tags'].append('shellinvocations')
			unpackreports[i]['tags'].append('shellinvocations')
			leaf_file = open(os.path.join(topleveldir, "filereports", "%s-filereport.pickle" % filehash), 'wb')
			cPickle.dump(leafreports, leaf_file)
			leaf_file.close()

## method to check if a file is an OpenSSH public or private key
## uses openssl to check
def checkOpenSSHKeys(filename, tags, blacklist=[], scanenv={}, scandebug=False, unpacktempdir=None):
	if not 'text' in tags:
		return

	opensshfile = open(filename, 'rb')
	opensshdata = opensshfile.readline()
	opensshfile.close()
	if "PRIVATE KEY" in opensshdata:
		## now run openssl
		p = subprocess.Popen(["openssl", "asn1parse", "-inform", "PEM", "-in", filename], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
		(stanout, stanerr) = p.communicate()
		if p.returncode != 0:
			return
		return (['privatekey'], None)

## stubs for cracking passwords with "<NAME>"
## 1. look for files called 'passwd' and 'shadow'
## 2. search for individual entries in the database
## 3. crack unknown entries
##
## This is implemented as an "aggregate scan". Running "<NAME>"
## is an expensive operation and often there can be duplicate password
## or shadow files in firmwares.
def crackPasswords(unpackreports, scantempdir, topleveldir, processors, scanenv, scandebug=False, unpacktempdir=None):
	passwdfiles = []
	for u in unpackreports.keys():
		if not (os.path.basename(u) == 'shadow' or os.path.basename(u) == 'passwd'):
			continue
		if 'symlink' in unpackreports[u]['tags']:
			continue
		if 'elf' in unpackreports[u]['tags']:
			continue
		if os.path.basename(u) == 'shadow':
{'fraction_non_alphanumeric': 0.12573627581873872, 'fraction_numerical': 0.006439959161234588, 'mean_word_length': 4.654529307282416, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2599173', 'n_tokens_mistral': 4580, 'n_tokens_neox': 4128, 'n_words': 1317}