~tribaal/txaws/xss-hardening

« back to all changes in this revision

Viewing changes to txaws/ec2/tests/test_client.py

  • Committer: Tristan Seligmann
  • Date: 2009-04-27 23:53:32 UTC
  • mfrom: (3.1.5 client)
  • Revision ID: mithrandi@mithrandi.net-20090427235332-l8azkl1kvn5pjtay
mergeĀ lp:~lifeless/txaws/client

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2009 Robert Collins <robertc@robertcollins.net>
 
2
# Licenced under the txaws licence available at /LICENSE in the txaws source.
 
3
 
 
4
import os
 
5
 
 
6
from twisted.internet.defer import succeed
 
7
 
 
8
from txaws.credentials import AWSCredentials
 
9
from txaws.ec2 import client
 
10
from txaws.tests import TXAWSTestCase
 
11
 
 
12
sample_describe_instances_result = """<?xml version="1.0"?>
 
13
<DescribeInstancesResponse xmlns="http://ec2.amazonaws.com/doc/2008-12-01/">
 
14
    <requestId>52b4c730-f29f-498d-94c1-91efb75994cc</requestId>
 
15
    <reservationSet>
 
16
        <item>
 
17
            <reservationId>r-cf24b1a6</reservationId>
 
18
            <ownerId>123456789012</ownerId>
 
19
            <groupSet>
 
20
                <item>
 
21
                    <groupId>default</groupId>
 
22
                </item>
 
23
            </groupSet>
 
24
            <instancesSet>
 
25
                <item>
 
26
                    <instanceId>i-abcdef01</instanceId>
 
27
                    <imageId>ami-12345678</imageId>
 
28
                    <instanceState>
 
29
                        <code>16</code>
 
30
                        <name>running</name>
 
31
                    </instanceState>
 
32
                    <privateDnsName>domU-12-31-39-03-15-11.compute-1.internal</privateDnsName>
 
33
                    <dnsName>ec2-75-101-245-65.compute-1.amazonaws.com</dnsName>
 
34
                    <reason/>
 
35
                    <keyName>keyname</keyName>
 
36
                    <amiLaunchIndex>0</amiLaunchIndex>
 
37
                    <productCodes/>
 
38
                    <instanceType>c1.xlarge</instanceType>
 
39
                    <launchTime>2009-04-27T02:23:18.000Z</launchTime>
 
40
                    <placement>
 
41
                        <availabilityZone>us-east-1c</availabilityZone>
 
42
                    </placement>
 
43
                    <kernelId>aki-b51cf9dc</kernelId>
 
44
                    <ramdiskId>ari-b31cf9da</ramdiskId>
 
45
                </item>
 
46
            </instancesSet>
 
47
        </item>
 
48
    </reservationSet>
 
49
</DescribeInstancesResponse>
 
50
"""
 
51
 
 
52
 
 
53
class TestEC2Client(TXAWSTestCase):
 
54
    
 
55
    def test_init_no_creds(self):
 
56
        os.environ['AWS_SECRET_ACCESS_KEY'] = 'foo'
 
57
        os.environ['AWS_ACCESS_KEY_ID'] = 'bar'
 
58
        ec2 = client.EC2Client()
 
59
        self.assertNotEqual(None, ec2.creds)
 
60
 
 
61
    def test_init_no_creds_non_available_errors(self):
 
62
        self.assertRaises(ValueError, client.EC2Client)
 
63
 
 
64
    def test_init_explicit_creds(self):
 
65
        creds = 'foo'
 
66
        ec2 = client.EC2Client(creds=creds)
 
67
        self.assertEqual(creds, ec2.creds)
 
68
 
 
69
    def test_describe_instances(self):
 
70
        class StubQuery(object):
 
71
            def __init__(stub, action, creds):
 
72
                self.assertEqual(action, 'DescribeInstances')
 
73
                self.assertEqual('foo', creds)
 
74
            def submit(self):
 
75
                return succeed(sample_describe_instances_result)
 
76
        ec2 = client.EC2Client(creds='foo', query_factory=StubQuery)
 
77
        d = ec2.describe_instances()
 
78
        def check_instances(reservation):
 
79
            self.assertEqual(1, len(reservation))
 
80
            self.assertEqual('i-abcdef01', reservation[0].instanceId)
 
81
            self.assertEqual('running', reservation[0].instanceState)
 
82
        d.addCallback(check_instances)
 
83
        return d
 
84
 
 
85
 
 
86
class TestQuery(TXAWSTestCase):
 
87
 
 
88
    def setUp(self):
 
89
        TXAWSTestCase.setUp(self)
 
90
        self.creds = AWSCredentials('foo', 'bar')
 
91
 
 
92
    def test_init_minimum(self):
 
93
        query = client.Query('DescribeInstances', self.creds)
 
94
        self.assertTrue('Timestamp' in query.params)
 
95
        del query.params['Timestamp']
 
96
        self.assertEqual(
 
97
            {'AWSAccessKeyId': 'foo',
 
98
             'Action': 'DescribeInstances',
 
99
             'SignatureMethod': 'HmacSHA1',
 
100
             'SignatureVersion': '2',
 
101
             'Version': '2008-12-01'},
 
102
            query.params)
 
103
 
 
104
    def test_init_requires_action(self):
 
105
        self.assertRaises(TypeError, client.Query)
 
106
 
 
107
    def test_init_requires_creds(self):
 
108
        self.assertRaises(TypeError, client.Query, None)
 
109
 
 
110
    def test_init_other_args_are_params(self):
 
111
        query = client.Query('DescribeInstances', self.creds,
 
112
            {'InstanceId.0': '12345'},
 
113
            time_tuple=(2007,11,12,13,14,15,0,0,0))
 
114
        self.assertEqual(
 
115
            {'AWSAccessKeyId': 'foo',
 
116
             'Action': 'DescribeInstances',
 
117
             'InstanceId.0': '12345',
 
118
             'SignatureMethod': 'HmacSHA1',
 
119
             'SignatureVersion': '2',
 
120
             'Timestamp': '2007-11-12T13:14:15Z',
 
121
             'Version': '2008-12-01'},
 
122
            query.params)
 
123
 
 
124
    def test_sorted_params(self):
 
125
        query = client.Query('DescribeInstances', self.creds,
 
126
            {'fun': 'games'},
 
127
            time_tuple=(2007,11,12,13,14,15,0,0,0))
 
128
        self.assertEqual([
 
129
            ('AWSAccessKeyId', 'foo'),
 
130
            ('Action', 'DescribeInstances'),
 
131
            ('SignatureMethod', 'HmacSHA1'),
 
132
            ('SignatureVersion', '2'),
 
133
            ('Timestamp', '2007-11-12T13:14:15Z'),
 
134
            ('Version', '2008-12-01'),
 
135
            ('fun', 'games'),
 
136
            ], query.sorted_params())
 
137
 
 
138
    def test_encode_unreserved(self):
 
139
        all_unreserved = ('ABCDEFGHIJKLMNOPQRSTUVWXYZ'
 
140
            'abcdefghijklmnopqrstuvwxyz0123456789-_.~')
 
141
        query = client.Query('DescribeInstances', self.creds)
 
142
        self.assertEqual(all_unreserved, query.encode(all_unreserved))
 
143
 
 
144
    def test_encode_space(self):
 
145
        """This may be just 'url encode', but the AWS manual isn't clear."""
 
146
        query = client.Query('DescribeInstances', self.creds)
 
147
        self.assertEqual('a%20space', query.encode('a space'))
 
148
 
 
149
    def test_canonical_query(self):
 
150
        query = client.Query('DescribeInstances', self.creds,
 
151
            {'fu n': 'g/ames', 'argwithnovalue':''},
 
152
            time_tuple=(2007,11,12,13,14,15,0,0,0))
 
153
        expected_query = ('AWSAccessKeyId=foo&Action=DescribeInstances'
 
154
            '&SignatureMethod=HmacSHA1&SignatureVersion=2&'
 
155
            'Timestamp=2007-11-12T13%3A14%3A15Z&Version=2008-12-01&'
 
156
            'argwithnovalue=&fu%20n=g%2Fames')
 
157
        self.assertEqual(expected_query, query.canonical_query_params())
 
158
 
 
159
    def test_signing_text(self):
 
160
        query = client.Query('DescribeInstances', self.creds,
 
161
            time_tuple=(2007,11,12,13,14,15,0,0,0))
 
162
        signing_text = ('GET\nec2.amazonaws.com\n/\n'
 
163
            'AWSAccessKeyId=foo&Action=DescribeInstances&'
 
164
            'SignatureMethod=HmacSHA1&SignatureVersion=2&'
 
165
            'Timestamp=2007-11-12T13%3A14%3A15Z&Version=2008-12-01')
 
166
        self.assertEqual(signing_text, query.signing_text())
 
167
 
 
168
    def test_sign(self):
 
169
        query = client.Query('DescribeInstances', self.creds,
 
170
            time_tuple=(2007,11,12,13,14,15,0,0,0))
 
171
        query.sign()
 
172
        self.assertEqual('4hEtLuZo9i6kuG3TOXvRQNOrE/U=',
 
173
            query.params['Signature'])