<s> """<STR_LIT>""" <EOL> from __future__ import print_function <EOL> from __future__ import absolute_import <EOL> from __future__ import division <EOL> from __future__ import unicode_literals <EOL> import os <EOL> import tempfile <EOL> import random <EOL> import time <EOL> import shutil <EOL> import json <EOL> import subprocess <EOL> import logging <EOL> import sys <EOL> if sys . version_info >= ( <NUM_LIT:2> , <NUM_LIT:7> ) : <EOL> import unittest <EOL> else : <EOL> import unittest2 as unittest <EOL> import tuf <EOL> import tuf . formats <EOL> import tuf . util <EOL> import tuf . log <EOL> import tuf . client . updater as updater <EOL> import tuf . unittest_toolbox as unittest_toolbox <EOL> import six <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> class TestArbitraryPackageAttack ( unittest_toolbox . Modified_TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . temporary_directory = tempfile . mkdtemp ( dir = os . getcwd ( ) ) <EOL> cls . SERVER_PORT = random . randint ( <NUM_LIT> , <NUM_LIT> ) <EOL> command = [ '<STR_LIT>' , '<STR_LIT>' , str ( cls . SERVER_PORT ) ] <EOL> cls . server_process = subprocess . Popen ( command , stderr = subprocess . PIPE ) <EOL> logger . info ( '<STR_LIT>' ) <EOL> logger . info ( '<STR_LIT>' + str ( cls . server_process . pid ) ) <EOL> logger . info ( '<STR_LIT>' + str ( cls . SERVER_PORT ) ) <EOL> cls . url = '<STR_LIT>' + str ( cls . SERVER_PORT ) + os . path . sep <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> shutil . rmtree ( cls . temporary_directory ) <EOL> if cls . server_process . returncode is None : <EOL> logger . info ( '<STR_LIT>' + str ( cls . server_process . pid ) + '<STR_LIT>' ) <EOL> cls . server_process . kill ( ) <EOL> def setUp ( self ) : <EOL> unittest_toolbox . Modified_TestCase . setUp ( self ) <EOL> original_repository_files = os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) <EOL> temporary_repository_root = self . make_temp_directory ( directory = self . temporary_directory ) <EOL> original_repository = os . path . join ( original_repository_files , '<STR_LIT>' ) <EOL> original_client = os . path . join ( original_repository_files , '<STR_LIT>' ) <EOL> self . repository_directory = os . path . join ( temporary_repository_root , '<STR_LIT>' ) <EOL> self . client_directory = os . path . join ( temporary_repository_root , '<STR_LIT>' ) <EOL> shutil . copytree ( original_repository , self . repository_directory ) <EOL> shutil . copytree ( original_client , self . client_directory ) <EOL> repository_basepath = self . repository_directory [ len ( os . getcwd ( ) ) : ] <EOL> url_prefix = '<STR_LIT>' + str ( self . SERVER_PORT ) + repository_basepath <EOL> tuf . conf . repository_directory = self . client_directory <EOL> self . repository_mirrors = { '<STR_LIT>' : { '<STR_LIT>' : url_prefix , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } } <EOL> self . repository_updater = updater . Updater ( '<STR_LIT>' , <EOL> self . repository_mirrors ) <EOL> def tearDown ( self ) : <EOL> unittest_toolbox . Modified_TestCase . tearDown ( self ) <EOL> def test_without_tuf ( self ) : <EOL> target_path = os . path . join ( self . repository_directory , '<STR_LIT>' , '<STR_LIT>' ) <EOL> client_target_path = os . path . join ( self . client_directory , '<STR_LIT>' ) <EOL> self . assertFalse ( os . path . exists ( client_target_path ) ) <EOL> length , hashes = tuf . util . get_file_details ( target_path ) <EOL> fileinfo = tuf . formats . make_fileinfo ( length , hashes ) <EOL> url_prefix = self . repository_mirrors [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> url_file = os . path . join ( url_prefix , '<STR_LIT>' , '<STR_LIT>' ) <EOL> six . moves . urllib . request . urlretrieve ( url_file , client_target_path ) <EOL> self . assertTrue ( os . path . exists ( client_target_path ) ) <EOL> length , hashes = tuf . util . get_file_details ( client_target_path ) <EOL> download_fileinfo = tuf . formats . make_fileinfo ( length , hashes ) <EOL> self . assertEqual ( fileinfo , download_fileinfo ) <EOL> with open ( target_path , '<STR_LIT>' ) as file_object : <EOL> file_object . write ( '<STR_LIT>' ) <EOL> length , hashes = tuf . util . get_file_details ( target_path ) <EOL> malicious_fileinfo = tuf . formats . make_fileinfo ( length , hashes ) <EOL> six . moves . urllib . request . urlretrieve ( url_file , client_target_path ) <EOL> length , hashes = tuf . util . get_file_details ( client_target_path ) <EOL> download_fileinfo = tuf . formats . make_fileinfo ( length , hashes ) <EOL> self . assertNotEqual ( download_fileinfo , fileinfo ) <EOL> self . assertEqual ( download_fileinfo , malicious_fileinfo ) <EOL> def test_with_tuf ( self ) : <EOL> file1_fileinfo = self . repository_updater . target ( '<STR_LIT>' ) <EOL> destination = os . path . join ( self . client_directory ) <EOL> self . repository_updater . download_target ( file1_fileinfo , destination ) <EOL> client_target_path = os . path . join ( destination , '<STR_LIT>' ) <EOL> self . assertTrue ( os . path . exists ( client_target_path ) ) <EOL> target_path = os . path . join ( self . repository_directory , '<STR_LIT>' , '<STR_LIT>' ) <EOL> with open ( target_path , '<STR_LIT>' ) as file_object : <EOL> file_object . write ( '<STR_LIT>' ) <EOL> try : <EOL> self . repository_updater . download_target ( file1_fileinfo , destination ) <EOL> except tuf . NoWorkingMirrorError as exception : <EOL> url_prefix = self . repository_mirrors [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> url_file = os . path . join ( url_prefix , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertTrue ( len ( exception . mirror_errors ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( url_file in exception . mirror_errors ) <EOL> self . assertTrue ( isinstance ( exception . mirror_errors [ url_file ] , <EOL> tuf . DownloadLengthMismatchError ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_with_tuf_and_metadata_tampering ( self ) : <EOL> target_path = os . path . join ( self . repository_directory , '<STR_LIT>' , '<STR_LIT>' ) <EOL> with open ( target_path , '<STR_LIT>' ) as file_object : <EOL> file_object . write ( '<STR_LIT>' ) <EOL> length , hashes = tuf . util . get_file_details ( target_path ) <EOL> metadata_path = os . path . join ( self . repository_directory , '<STR_LIT>' , '<STR_LIT>' ) <EOL> metadata = tuf . util . load_json_file ( metadata_path ) <EOL> metadata [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] = hashes <EOL> metadata [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] = length <EOL> tuf . formats . check_signable_object_format ( metadata ) <EOL> with open ( metadata_path , '<STR_LIT:wb>' ) as file_object : <EOL> json . dumps ( metadata , file_object , indent = <NUM_LIT:1> , sort_keys = True ) . encode ( '<STR_LIT:utf-8>' ) <EOL> try : <EOL> self . repository_updater . refresh ( ) <EOL> file1_fileinfo = self . repository_updater . target ( '<STR_LIT>' ) <EOL> destination = os . path . join ( self . client_directory ) <EOL> self . repository_updater . download_target ( file1_fileinfo , destination ) <EOL> except tuf . NoWorkingMirrorError as exception : <EOL> url_prefix = self . repository_mirrors [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> url_file = os . path . join ( url_prefix , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertTrue ( len ( exception . mirror_errors ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( url_file in exception . mirror_errors ) <EOL> self . assertTrue ( isinstance ( exception . mirror_errors [ url_file ] , <EOL> tuf . DownloadLengthMismatchError ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
